rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(crm): persist contact data imports

Browse Source
This commit is contained in:
rogee
2026-06-05 03:20:34 +08:00
parent ab2ca840fb
commit 8eed2f1943
10 changed files with 289 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/handler/api/v1/contact_handler.go
+3 -3
View File
@@ -724,18 +724,18 @@ func (h *ContactHandler) Import(c *gin.Context) {
file, _, fileErr = c.Request.FormFile("file")
}
if fileErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": "csv file required"})
c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "failed to import contacts"})
return
}
defer file.Close()
result, svcErr := h.svc.ImportCSV(c.Request.Context(), accountID, file)
_, svcErr := h.svc.ImportContacts(c.Request.Context(), accountID, getUserID(c), file)
if svcErr != nil {
c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "failed to import contacts"})
return
}
c.JSON(http.StatusOK, result)
c.Status(http.StatusOK)
}
// ContactableInboxes returns inboxes that a contact can be associated with.
internal/handler/api/v1/contact_handler_crud_test.go
+38
View File
@@ -4,6 +4,7 @@ import (
"bytes"
"encoding/json"
"fmt"
"mime/multipart"
"net/http"
"net/http/httptest"
"testing"
@@ -51,6 +52,7 @@ func (s *ContactHandlerCRUDTestSuite) SetupSuite() {
&model.Contact{},
&model.Tag{},
&model.ContactLabel{},
&model.DataImport{},
&model.Conversation{},
&model.Message{},
&model.ContactInbox{},
@@ -82,6 +84,7 @@ func (s *ContactHandlerCRUDTestSuite) SetupSuite() {
s.router.Use(gin.Recovery(), s.mockAuthMiddleware())
s.router.GET("/api/v1/accounts/:id/contacts", s.handler.List)
s.router.GET("/api/v1/accounts/:id/contacts/search", s.handler.Search)
s.router.POST("/api/v1/accounts/:id/contacts/import", s.handler.Import)
s.router.GET("/api/v1/accounts/:id/contacts/:contact_id", s.handler.Get)
s.router.POST("/api/v1/accounts/:id/contacts", s.handler.Create)
s.router.PUT("/api/v1/accounts/:id/contacts/:contact_id", s.handler.Update)
@@ -120,6 +123,7 @@ func (s *ContactHandlerCRUDTestSuite) SetupTest() {
s.db.Exec("DELETE FROM contact_notes")
s.db.Exec("DELETE FROM contact_labels")
s.db.Exec("DELETE FROM tags")
s.db.Exec("DELETE FROM data_imports")
s.db.Exec("DELETE FROM messages")
s.db.Exec("DELETE FROM notes")
s.db.Exec("DELETE FROM contact_inboxes")
@@ -597,6 +601,40 @@ func (s *ContactHandlerCRUDTestSuite) TestMerge_ChatwootActionsPathReturnsRawCon
s.Equal(int64(1), count)
}
func (s *ContactHandlerCRUDTestSuite) TestImport_CreatesDataImportAndReturnsOK() {
var body bytes.Buffer
writer := multipart.NewWriter(&body)
part, err := writer.CreateFormFile("import_file", "contacts.csv")
s.Require().NoError(err)
_, err = part.Write([]byte("name,email,labels\nImported,imported@example.com,vip\n"))
s.Require().NoError(err)
s.Require().NoError(writer.Close())
w := httptest.NewRecorder()
req, _ := http.NewRequest("POST",
fmt.Sprintf("/api/v1/accounts/%d/contacts/import", s.account.ID),
&body)
req.Header.Set("Content-Type", writer.FormDataContentType())
s.router.ServeHTTP(w, req)
s.Equal(http.StatusOK, w.Code)
s.Empty(w.Body.String())
var dataImport model.DataImport
s.Require().NoError(s.db.Where("account_id = ? AND data_type = ?", s.account.ID, "contacts").First(&dataImport).Error)
s.Equal(string(model.DataImportStatusCompleted), dataImport.Status)
s.Equal(1, dataImport.TotalRecords)
s.Equal(1, dataImport.ProcessedRecords)
s.Require().NotNil(dataImport.UserID)
s.Equal(s.user.ID, *dataImport.UserID)
var contact model.Contact
s.Require().NoError(s.db.Where("account_id = ? AND email = ?", s.account.ID, "imported@example.com").First(&contact).Error)
var labelCount int64
s.db.Model(&model.ContactLabel{}).Where("account_id = ? AND contact_id = ?", s.account.ID, contact.ID).Count(&labelCount)
s.Equal(int64(1), labelCount)
}
func (s *ContactHandlerCRUDTestSuite) TestLabels_UpdateListAndFilter() {
bodyBytes, _ := json.Marshal(map[string]interface{}{"labels": []string{"vip", "trial"}})
internal/handler/api/v1/contact_handler_g3_test.go
+3 -3
View File
@@ -89,8 +89,8 @@ func TestContactImportMissingFile(t *testing.T) {
req, _ := http.NewRequest("POST", "/api/v1/accounts/1/contacts/import", nil)
router.ServeHTTP(w, req)
// Without a multipart form, FormFile will error, so we expect 400
assert.Equal(t, http.StatusBadRequest, w.Code)
// Chatwoot returns 422 when import_file is missing.
assert.Equal(t, http.StatusUnprocessableEntity, w.Code)
var resp map[string]interface{}
json.Unmarshal(w.Body.Bytes(), &resp)
assert.Contains(t, resp, "error")
@@ -138,4 +138,4 @@ func TestDeleteCustomAttributesBadContactID(t *testing.T) {
router.ServeHTTP(w, req)
assert.Equal(t, http.StatusBadRequest, w.Code)
}
}
internal/model/data_import.go
+14 -13
View File
@@ -10,21 +10,22 @@ import (
// DataImport represents a bulk data import task.
// Reference: Chatwoot DataImport model + P2B M12 spec
type DataImport struct {
ID uint `gorm:"primaryKey" json:"id"`
AccountID uint `gorm:"not null;index" json:"account_id"`
UserID uint `gorm:"not null" json:"user_id"`
SourceType string `gorm:"size:100;not null" json:"source_type"` // csv/chatwoot/zendesk/freshdesk
Status string `gorm:"size:50;not null;default:'pending'" json:"status"` // pending/processing/completed/failed
TotalRecords int `json:"total_records"`
ProcessedRecords int `json:"processed_records"`
FailedRecords int `json:"failed_records"`
ImportConfig json.RawMessage `gorm:"type:jsonb" json:"import_config"`
CreatedAt time.Time `gorm:"autoCreateTime" json:"created_at"`
UpdatedAt time.Time `gorm:"autoUpdateTime" json:"updated_at"`
DeletedAt gorm.DeletedAt `gorm:"index" json:"deleted_at,omitempty"`
ID uint `gorm:"primaryKey" json:"id"`
AccountID uint `gorm:"not null;index" json:"account_id"`
UserID *uint `gorm:"index" json:"user_id,omitempty"`
DataType string `gorm:"size:100;not null" json:"data_type"` // Chatwoot currently supports contacts
Status string `gorm:"size:50;not null;default:'pending'" json:"status"` // pending/processing/completed/failed
TotalRecords int `json:"total_records"`
ProcessedRecords int `json:"processed_records"`
FailedRecords int `json:"failed_records"`
ProcessingErrors string `gorm:"type:text" json:"processing_errors,omitempty"`
ImportConfig json.RawMessage `gorm:"type:jsonb" json:"import_config"`
CreatedAt time.Time `gorm:"autoCreateTime" json:"created_at"`
UpdatedAt time.Time `gorm:"autoUpdateTime" json:"updated_at"`
DeletedAt gorm.DeletedAt `gorm:"index" json:"deleted_at,omitempty"`
Account Account `gorm:"foreignKey:AccountID" json:"account,omitempty"`
User User `gorm:"foreignKey:UserID" json:"user,omitempty"`
}
func (DataImport) TableName() string { return "data_imports" }
func (DataImport) TableName() string { return "data_imports" }
internal/repository/testdb_helper.go
+1
View File
@@ -152,6 +152,7 @@ func defaultTestModels() []interface{} {
&model.Tag{},
&model.ConversationLabel{},
&model.ContactLabel{},
&model.DataImport{},
&model.PlatformApp{},
&model.Permissible{},
&model.AccessToken{},
internal/service/contact_service.go
+171 -17
View File
@@ -350,6 +350,45 @@ type ImportCSVResult struct {
Failed int `json:"failed"`
}
func (s *ContactService) ImportContacts(ctx context.Context, accountID, userID uint, r io.Reader) (*model.DataImport, error) {
if !s.Ready() {
return nil, errors.New("contact service not ready")
}
var userIDPtr *uint
if userID != 0 {
userIDPtr = &userID
}
dataImport := &model.DataImport{AccountID: accountID, UserID: userIDPtr, DataType: "contacts", Status: string(model.DataImportStatusPending)}
if err := s.repo.DB().WithContext(ctx).Create(dataImport).Error; err != nil {
return nil, err
}
if err := s.repo.DB().WithContext(ctx).Model(dataImport).Update("status", string(model.DataImportStatusProcessing)).Error; err != nil {
return nil, err
}
result, err := s.ImportCSV(ctx, accountID, r)
if err != nil {
s.repo.DB().WithContext(ctx).Model(dataImport).Updates(map[string]any{
"status": string(model.DataImportStatusFailed),
"processing_errors": err.Error(),
})
return dataImport, err
}
updates := map[string]any{
"status": string(model.DataImportStatusCompleted),
"processed_records": result.Imported,
"failed_records": result.Failed,
"total_records": result.Imported + result.Skipped + result.Failed,
}
if err := s.repo.DB().WithContext(ctx).Model(dataImport).Updates(updates).Error; err != nil {
return nil, err
}
if err := s.repo.DB().WithContext(ctx).First(dataImport, dataImport.ID).Error; err != nil {
return nil, err
}
return dataImport, nil
}
// ImportCSV reads contacts from a CSV reader and creates them.
// POST /api/v1/accounts/:id/contacts/import
// Reference: Chatwoot contacts#import
@@ -379,9 +418,9 @@ func (s *ContactService) ImportCSV(ctx context.Context, accountID uint, r io.Rea
continue
}
contact := &model.Contact{
AccountID: accountID,
}
contact := &model.Contact{AccountID: accountID}
customAttributes := map[string]any{}
var labels []string
if idx, ok := colIndex["name"]; ok && idx < len(row) {
contact.Name = row[idx]
@@ -390,7 +429,7 @@ func (s *ContactService) ImportCSV(ctx context.Context, accountID uint, r io.Rea
contact.Email = row[idx]
}
if idx, ok := colIndex["phone_number"]; ok && idx < len(row) {
contact.PhoneNumber = row[idx]
contact.PhoneNumber = formatImportPhone(row[idx])
}
if idx, ok := colIndex["identifier"]; ok && idx < len(row) {
contact.Identifier = row[idx]
@@ -401,26 +440,48 @@ func (s *ContactService) ImportCSV(ctx context.Context, accountID uint, r io.Rea
if idx, ok := colIndex["location"]; ok && idx < len(row) {
contact.Location = row[idx]
}
if idx, ok := colIndex["city"]; ok && idx < len(row) && row[idx] != "" {
contact.Location = row[idx]
}
if idx, ok := colIndex["company_name"]; ok && idx < len(row) && row[idx] != "" {
customAttributes["company_name"] = row[idx]
}
if idx, ok := colIndex["contact_type"]; ok && idx < len(row) {
contact.ContactType = row[idx]
}
if idx, ok := colIndex["labels"]; ok && idx < len(row) {
labels = splitImportLabels(row[idx])
}
known := map[string]struct{}{"name": {}, "email": {}, "phone_number": {}, "identifier": {}, "country_code": {}, "location": {}, "city": {}, "company_name": {}, "contact_type": {}, "labels": {}}
for i, col := range header {
col = strings.TrimSpace(col)
if col == "" || i >= len(row) {
continue
}
if _, ok := known[col]; ok || row[i] == "" {
continue
}
customAttributes[col] = row[i]
}
// Skip rows without name (required field)
if contact.Name == "" {
result.Skipped++
existing := s.findImportContact(ctx, accountID, contact)
if existing != nil {
mergeImportContact(existing, contact, customAttributes)
if err := s.repo.Update(ctx, existing); err != nil {
applogger.L().Errorf("Failed to update imported contact row: %v", err)
result.Failed++
continue
}
if err := s.updateImportedLabels(ctx, accountID, existing.ID, labels); err != nil {
applogger.L().Errorf("Failed to update imported contact labels: %v", err)
result.Failed++
continue
}
result.Imported++
continue
}
// Check for duplicate by email
if contact.Email != "" {
existing, _ := s.repo.FindByEmail(ctx, accountID, contact.Email)
if existing != nil {
result.Skipped++
continue
}
}
contact.CustomAttributes = datatypes.JSON("{}")
contact.CustomAttributes = jsonFromMap(customAttributes)
contact.AdditionalAttributes = datatypes.JSON("{}")
if err := s.repo.Create(ctx, contact); err != nil {
@@ -428,12 +489,105 @@ func (s *ContactService) ImportCSV(ctx context.Context, accountID uint, r io.Rea
result.Failed++
continue
}
if err := s.updateImportedLabels(ctx, accountID, contact.ID, labels); err != nil {
applogger.L().Errorf("Failed to update imported contact labels: %v", err)
result.Failed++
continue
}
result.Imported++
}
return result, nil
}
func (s *ContactService) findImportContact(ctx context.Context, accountID uint, contact *model.Contact) *model.Contact {
if contact.Identifier != "" {
if existing, err := s.repo.FindByIdentifier(ctx, accountID, contact.Identifier); err == nil {
return existing
}
}
if contact.Email != "" {
if existing, err := s.repo.FindByEmail(ctx, accountID, contact.Email); err == nil {
return existing
}
}
if contact.PhoneNumber != "" {
var existing model.Contact
if err := s.repo.DB().WithContext(ctx).Where("account_id = ? AND phone_number = ?", accountID, contact.PhoneNumber).First(&existing).Error; err == nil {
return &existing
}
}
return nil
}
func mergeImportContact(existing *model.Contact, incoming *model.Contact, customAttributes map[string]any) {
if incoming.Identifier != "" {
existing.Identifier = incoming.Identifier
}
if incoming.Email != "" {
existing.Email = incoming.Email
}
if incoming.PhoneNumber != "" {
existing.PhoneNumber = incoming.PhoneNumber
}
if incoming.Name != "" {
existing.Name = incoming.Name
}
if incoming.CountryCode != "" {
existing.CountryCode = incoming.CountryCode
}
if incoming.Location != "" {
existing.Location = incoming.Location
}
if incoming.ContactType != "" {
existing.ContactType = incoming.ContactType
}
merged := map[string]any{}
if len(existing.CustomAttributes) > 0 {
_ = json.Unmarshal(existing.CustomAttributes, &merged)
}
for key, value := range customAttributes {
merged[key] = value
}
existing.CustomAttributes = jsonFromMap(merged)
}
func jsonFromMap(values map[string]any) datatypes.JSON {
if len(values) == 0 {
return datatypes.JSON("{}")
}
bytes, _ := json.Marshal(values)
return datatypes.JSON(bytes)
}
func formatImportPhone(phone string) string {
phone = strings.TrimSpace(phone)
if phone == "" || strings.HasPrefix(phone, "+") {
return phone
}
return "+" + phone
}
func splitImportLabels(raw string) []string {
parts := strings.Split(raw, ",")
labels := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part != "" {
labels = append(labels, part)
}
}
return labels
}
func (s *ContactService) updateImportedLabels(ctx context.Context, accountID, contactID uint, labels []string) error {
if len(labels) == 0 {
return nil
}
_, err := s.UpdateLabels(ctx, accountID, contactID, labels)
return err
}
// Filter retrieves contacts matching advanced filter criteria.
// POST /api/v1/accounts/:id/contacts/filter
// Reference: Chatwoot ContactFilterService#perform — filters by contact_type, source,
internal/service/contact_service_g3_test.go
+35 -10
View File
@@ -204,19 +204,19 @@ func TestContactService_ImportCSV_ImportsValidRows(t *testing.T) {
assert.Equal(t, int64(2), count)
}
func TestContactService_ImportCSV_SkipsRowsWithoutName(t *testing.T) {
func TestContactService_ImportCSV_AllowsRowsWithoutNameWhenIdentityPresent(t *testing.T) {
db, _, svc := setupContactService(t)
account := createTestAccount(t, db)
csvData := "name,email\nAlice,alice@test.com\n,bob@test.com\n"
result, err := svc.ImportCSV(context.Background(), account.ID, strings.NewReader(csvData))
require.NoError(t, err)
assert.Equal(t, 1, result.Imported)
assert.Equal(t, 1, result.Skipped) // row without name is skipped
assert.Equal(t, 2, result.Imported)
assert.Equal(t, 0, result.Skipped)
assert.Equal(t, 0, result.Failed)
}
func TestContactService_ImportCSV_SkipsDuplicateEmail(t *testing.T) {
func TestContactService_ImportCSV_MergesDuplicateEmail(t *testing.T) {
db, _, svc := setupContactService(t)
account := createTestAccount(t, db)
@@ -231,9 +231,34 @@ func TestContactService_ImportCSV_SkipsDuplicateEmail(t *testing.T) {
csvData := "name,email\nNewDup,dup@test.com\nUnique,unique@test.com\n"
result, err := svc.ImportCSV(context.Background(), account.ID, strings.NewReader(csvData))
require.NoError(t, err)
assert.Equal(t, 1, result.Imported) // unique@test.com
assert.Equal(t, 1, result.Skipped) // dup@test.com is skipped
assert.Equal(t, 2, result.Imported)
assert.Equal(t, 0, result.Skipped)
assert.Equal(t, 0, result.Failed)
require.NoError(t, db.First(existing, existing.ID).Error)
assert.Equal(t, "NewDup", existing.Name)
var count int64
db.Model(&model.Contact{}).Where("account_id = ?", account.ID).Count(&count)
assert.Equal(t, int64(2), count)
}
func TestContactService_ImportContacts_CreatesCompletedDataImport(t *testing.T) {
db, _, svc := setupContactService(t)
account := createTestAccount(t, db)
userID := uint(42)
csvData := "name,email,labels\nAlice,alice@test.com,vip\n"
dataImport, err := svc.ImportContacts(context.Background(), account.ID, userID, strings.NewReader(csvData))
require.NoError(t, err)
assert.Equal(t, "contacts", dataImport.DataType)
assert.Equal(t, string(model.DataImportStatusCompleted), dataImport.Status)
assert.Equal(t, 1, dataImport.TotalRecords)
assert.Equal(t, 1, dataImport.ProcessedRecords)
assert.NotNil(t, dataImport.UserID)
assert.Equal(t, userID, *dataImport.UserID)
var labelCount int64
db.Model(&model.ContactLabel{}).Where("account_id = ?", account.ID).Count(&labelCount)
assert.Equal(t, int64(1), labelCount)
}
func TestContactService_ImportCSV_FailsOnBadCSV(t *testing.T) {
@@ -274,9 +299,9 @@ func TestContactService_DeleteCustomAttributes_ClearsCustomAttributes(t *testing
// Create a contact with custom attributes
contact := &model.Contact{
AccountID: account.ID,
Name: "Custom Attr Contact",
Email: "custom@test.com",
AccountID: account.ID,
Name: "Custom Attr Contact",
Email: "custom@test.com",
CustomAttributes: datatypes.JSON(`{"key1":"val1","key2":"val2"}`),
}
require.NoError(t, db.Create(contact).Error)
@@ -416,4 +441,4 @@ func TestContactService_GetContactableInboxes_NoInboxes(t *testing.T) {
result, err := svc.GetContactableInboxes(context.Background(), account.ID, contact.ID)
require.NoError(t, err)
assert.Len(t, result, 0)
}
}
internal/service/service_test_helper.go
+4
View File
@@ -68,6 +68,7 @@ func setupServiceTestDB(t *testing.T) *gorm.DB {
&model.CompanyNote{},
&model.Tag{},
&model.ContactLabel{},
&model.DataImport{},
); err != nil {
t.Fatalf("failed to auto-migrate models: %v", err)
}
@@ -270,6 +271,9 @@ func setupContactServiceTestDB(t *testing.T) *gorm.DB {
&model.Inbox{},
&model.Contact{},
&model.ContactInbox{},
&model.Tag{},
&model.ContactLabel{},
&model.DataImport{},
); err != nil {
t.Fatalf("failed to auto-migrate contact models: %v", err)
}
migrations/000020_add_data_imports.down.sql
+1
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS data_imports;
migrations/000020_add_data_imports.up.sql
+19
View File
@@ -0,0 +1,19 @@
CREATE TABLE IF NOT EXISTS data_imports (
id INTEGER PRIMARY KEY,
account_id INTEGER NOT NULL,
user_id INTEGER,
data_type VARCHAR(100) NOT NULL,
status VARCHAR(50) NOT NULL DEFAULT 'pending',
total_records INTEGER NOT NULL DEFAULT 0,
processed_records INTEGER NOT NULL DEFAULT 0,
failed_records INTEGER NOT NULL DEFAULT 0,
processing_errors TEXT,
import_config JSONB DEFAULT '{}',
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
deleted_at TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_data_imports_account_id ON data_imports(account_id);
CREATE INDEX IF NOT EXISTS idx_data_imports_deleted_at ON data_imports(deleted_at);
CREATE INDEX IF NOT EXISTS idx_data_imports_user_id ON data_imports(user_id);