feat: create CRM tasks from campaign actions

internal/app/advanced/service.go

@@ -53,6 +53,12 @@ type Service interface {
 	ProcessIncomingReply(ctx context.Context, emailAccountID uuid.UUID, msg *models.EmailMessageStoreData) *errx.Error
 	GetABWinnerAnalysis(ctx context.Context, organizationID, campaignID uuid.UUID) (*models.ABWinnerAnalysis, *errx.Error)
 
+	// CreateContactTask creates a CRM task for a contact, used by the campaign
+	// "create task" action node. createdBy is the campaign owner; the task's
+	// AssignedTo (set in data) is the teammate chosen on the step. Records a
+	// task_created activity on the contact.
+	CreateContactTask(ctx context.Context, orgID, createdBy uuid.UUID, data *models.CreateCRMTask) (*models.CRMTask, *errx.Error)
+
 	// WireDispatcher attaches the event dispatcher that fans classified
 	// replies + deliverability events out to customer webhooks and third-party
 	// integration actions (Slack ping, CRM upsert).
@@ -281,6 +287,24 @@ func (s *service) ShouldSuppressRecipient(ctx context.Context, organizationID uu
 // Unsubscribe resolves the campaign + contact behind a List-Unsubscribe link and
 // suppresses the recipient org-wide. Always suppresses (an explicit recipient
 // request), then fans out the campaign.unsubscribed event for Slack/CRM.
+func (s *service) CreateContactTask(ctx context.Context, orgID, createdBy uuid.UUID, data *models.CreateCRMTask) (*models.CRMTask, *errx.Error) {
+	if s.crmRepo == nil {
+		return nil, errx.InternalError()
+	}
+	task, err := s.crmRepo.CreateCRMTask(ctx, orgID, createdBy, data)
+	if err != nil {
+		return nil, errx.InternalError()
+	}
+	if task.ContactID != nil {
+		_ = s.crmRepo.RecordActivity(ctx, orgID, *task.ContactID, &createdBy, models.ActivityTaskCreated, map[string]interface{}{
+			"task_id":    task.ID.String(),
+			"task_title": task.Title,
+			"source":     "campaign",
+		})
+	}
+	return task, nil
+}
+
 func (s *service) Unsubscribe(ctx context.Context, campaignID, contactID uuid.UUID) *errx.Error {
 	campaign, err := s.campaignRepo.GetByID(ctx, campaignID)
 	if err != nil || campaign == nil || campaign.OrganizationID == nil {

internal/models/sequence.go

@@ -42,7 +42,7 @@ type Sequence struct {
 // ActionConfig is the persisted config for a non-email (action/wait) node. Type
 // is the switch the task executes on; the remaining fields are type-scoped.
 type ActionConfig struct {
-	Type string `json:"type"` // wait | add_tag | remove_tag | unsubscribe | notify | end
+	Type string `json:"type"` // wait | add_tag | remove_tag | unsubscribe | notify | create_task | end
 
 	// wait
 	WaitMinutes *int `json:"wait_minutes,omitempty"`
@@ -53,6 +53,15 @@ type ActionConfig struct {
 	// notify — webhook / integration fan-out
 	NotifyEvent string         `json:"notify_event,omitempty"`
 	NotifyData  map[string]any `json:"notify_data,omitempty"`
+
+	// create_task — open a CRM task for the lead when they reach this step
+	// (e.g. a Call task). TaskAssignedTo is the teammate chosen on the step;
+	// when nil the task falls back to the campaign owner.
+	TaskTitle         string     `json:"task_title,omitempty"`
+	TaskType          string     `json:"task_type,omitempty"`     // general | call | email | meeting
+	TaskPriority      string     `json:"task_priority,omitempty"` // low | medium | high | urgent
+	TaskAssignedTo    *uuid.UUID `json:"task_assigned_to,omitempty"`
+	TaskDueOffsetDays *int       `json:"task_due_offset_days,omitempty"` // due N days after the step fires
 }
 
 type UpdateSequence struct {

internal/repository/sequence_actions.go

@@ -14,7 +14,7 @@ func validateActionConfig(a *models.ActionConfig) *errx.Error {
 		return nil
 	}
 	switch a.Type {
-	case "wait", "add_tag", "remove_tag", "unsubscribe", "notify", "end":
+	case "wait", "add_tag", "remove_tag", "unsubscribe", "notify", "create_task", "end":
 		// Type must be known. Sub-config (wait minutes, tag) is filled in the
 		// editor; an unconfigured node is a harmless no-op at send time, so we
 		// don't block creating a draft node here.

internal/tasks/campaign_task.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/getsentry/sentry-go"
@@ -736,6 +737,45 @@ func (s *tasksService) executeActionNode(ctx context.Context, campaign *models.C
 		}
 		s.advanced.EmitCampaignEvent(ctx, *campaign.OrganizationID, event, data)
 		return nil
+	case "create_task":
+		if s.advanced == nil || campaign.OrganizationID == nil {
+			return nil
+		}
+		owner, perr := uuid.Parse(campaign.UserID)
+		if perr != nil {
+			return nil
+		}
+		title := strings.TrimSpace(cfg.TaskTitle)
+		if title == "" {
+			name := strings.TrimSpace(contact.FirstName + " " + contact.LastName)
+			if name == "" {
+				name = contact.Email
+			}
+			title = "Follow up: " + name
+		}
+		// Per-step assignee; fall back to the campaign owner when unset.
+		assignee := cfg.TaskAssignedTo
+		if assignee == nil {
+			assignee = &owner
+		}
+		// Task types are user-managed free text; pass the configured name
+		// through (empty = untyped).
+		cid := contact.ID
+		data := &models.CreateCRMTask{
+			ContactID:  &cid,
+			Title:      title,
+			Type:       cfg.TaskType,
+			Priority:   cfg.TaskPriority,
+			AssignedTo: assignee,
+		}
+		if cfg.TaskDueOffsetDays != nil {
+			due := time.Now().UTC().AddDate(0, 0, *cfg.TaskDueOffsetDays)
+			data.DueDate = &due
+		}
+		if _, xerr := s.advanced.CreateContactTask(ctx, *campaign.OrganizationID, owner, data); xerr != nil {
+			return xerr
+		}
+		return nil
 	default:
 		return nil
 	}