diff --git a/internal/handler/api/v1/contact_handler.go b/internal/handler/api/v1/contact_handler.go index 55f61676..2bbf609d 100644 --- a/internal/handler/api/v1/contact_handler.go +++ b/internal/handler/api/v1/contact_handler.go @@ -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. diff --git a/internal/handler/api/v1/contact_handler_crud_test.go b/internal/handler/api/v1/contact_handler_crud_test.go index 6b90d5c2..5f432879 100644 --- a/internal/handler/api/v1/contact_handler_crud_test.go +++ b/internal/handler/api/v1/contact_handler_crud_test.go @@ -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"}}) diff --git a/internal/handler/api/v1/contact_handler_g3_test.go b/internal/handler/api/v1/contact_handler_g3_test.go index f5b55cfc..40449f75 100644 --- a/internal/handler/api/v1/contact_handler_g3_test.go +++ b/internal/handler/api/v1/contact_handler_g3_test.go @@ -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) -} \ No newline at end of file +} diff --git a/internal/model/data_import.go b/internal/model/data_import.go index de6f69ef..deef40db 100644 --- a/internal/model/data_import.go +++ b/internal/model/data_import.go @@ -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" } \ No newline at end of file +func (DataImport) TableName() string { return "data_imports" } diff --git a/internal/repository/testdb_helper.go b/internal/repository/testdb_helper.go index 77c39d73..5f008834 100644 --- a/internal/repository/testdb_helper.go +++ b/internal/repository/testdb_helper.go @@ -152,6 +152,7 @@ func defaultTestModels() []interface{} { &model.Tag{}, &model.ConversationLabel{}, &model.ContactLabel{}, + &model.DataImport{}, &model.PlatformApp{}, &model.Permissible{}, &model.AccessToken{}, diff --git a/internal/service/contact_service.go b/internal/service/contact_service.go index be0bf701..18b49c9a 100644 --- a/internal/service/contact_service.go +++ b/internal/service/contact_service.go @@ -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, diff --git a/internal/service/contact_service_g3_test.go b/internal/service/contact_service_g3_test.go index 878023c2..4f5a54f7 100644 --- a/internal/service/contact_service_g3_test.go +++ b/internal/service/contact_service_g3_test.go @@ -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) -} \ No newline at end of file +} diff --git a/internal/service/service_test_helper.go b/internal/service/service_test_helper.go index c77b1275..8452ac81 100644 --- a/internal/service/service_test_helper.go +++ b/internal/service/service_test_helper.go @@ -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) } diff --git a/migrations/000020_add_data_imports.down.sql b/migrations/000020_add_data_imports.down.sql new file mode 100644 index 00000000..d2ea4bd8 --- /dev/null +++ b/migrations/000020_add_data_imports.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS data_imports; diff --git a/migrations/000020_add_data_imports.up.sql b/migrations/000020_add_data_imports.up.sql new file mode 100644 index 00000000..e85889af --- /dev/null +++ b/migrations/000020_add_data_imports.up.sql @@ -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);