feat(api): add CSV bulk importer for project work items

apps/api/cmd/api/main.go

@@ -95,7 +95,7 @@ func main() {
 		mc = client
 	}
 
-	r := router.New(router.Config{
+	r, importerSvc := router.New(router.Config{
 		Log:               log,
 		DB:                db,
 		Redis:             rdb,
@@ -120,7 +120,8 @@ func main() {
 			consumer.Register(queue.QueueEmails, queue.HandleSendEmail(log, emailSender))
 			webhookDeliverer := service.NewWebhookDeliverer(store.NewWebhookStore(db), log)
 			consumer.Register(queue.QueueWebhooks, queue.HandleWebhook(webhookDeliverer))
-			if err := consumer.Run(consumerCtx, []string{queue.QueueEmails, queue.QueueWebhooks}); err != nil {
+			consumer.Register(queue.QueueImports, queue.HandleImport(importerSvc.Run))
+			if err := consumer.Run(consumerCtx, []string{queue.QueueEmails, queue.QueueWebhooks, queue.QueueImports}); err != nil {
 				log.Warn("queue consumer", "error", err)
 			}
 		}

apps/api/internal/handler/importer.go

@@ -0,0 +1,117 @@
+package handler
+
+import (
+	"errors"
+	"net/http"
+
+	"github.com/Devlaner/devlane/api/internal/middleware"
+	"github.com/Devlaner/devlane/api/internal/service"
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+)
+
+// maxImportUpload caps the CSV upload size (10 MiB).
+const maxImportUpload = 10 << 20
+
+// ImporterHandler serves project bulk-import (CSV) creation + status.
+type ImporterHandler struct {
+	Importers *service.ImporterService
+}
+
+// Create accepts a multipart CSV upload and starts an import.
+// POST /api/workspaces/:slug/projects/:projectId/importers/
+func (h *ImporterHandler) Create(c *gin.Context) {
+	user := middleware.GetUser(c)
+	if user == nil {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "Authentication required"})
+		return
+	}
+	projectID, err := uuid.Parse(c.Param("projectId"))
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid project ID"})
+		return
+	}
+	fileHeader, err := c.FormFile("file")
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "A CSV file is required (field 'file')"})
+		return
+	}
+	if fileHeader.Size > maxImportUpload {
+		c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": "The CSV file is too large"})
+		return
+	}
+	f, err := fileHeader.Open()
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Could not read the uploaded file"})
+		return
+	}
+	defer f.Close()
+
+	im, err := h.Importers.CreateCSV(c.Request.Context(), c.Param("slug"), projectID, user.ID, fileHeader.Filename, f)
+	if err != nil {
+		h.importError(c, err)
+		return
+	}
+	c.JSON(http.StatusCreated, im)
+}
+
+// List returns a project's imports.
+// GET /api/workspaces/:slug/projects/:projectId/importers/
+func (h *ImporterHandler) List(c *gin.Context) {
+	user := middleware.GetUser(c)
+	if user == nil {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "Authentication required"})
+		return
+	}
+	projectID, err := uuid.Parse(c.Param("projectId"))
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid project ID"})
+		return
+	}
+	list, err := h.Importers.List(c.Request.Context(), c.Param("slug"), projectID, user.ID)
+	if err != nil {
+		h.importError(c, err)
+		return
+	}
+	c.JSON(http.StatusOK, list)
+}
+
+// Get returns a single import's status.
+// GET /api/workspaces/:slug/projects/:projectId/importers/:importerId/
+func (h *ImporterHandler) Get(c *gin.Context) {
+	user := middleware.GetUser(c)
+	if user == nil {
+		c.JSON(http.StatusUnauthorized, gin.H{"error": "Authentication required"})
+		return
+	}
+	projectID, err := uuid.Parse(c.Param("projectId"))
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid project ID"})
+		return
+	}
+	id, err := uuid.Parse(c.Param("importerId"))
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid import ID"})
+		return
+	}
+	im, err := h.Importers.Get(c.Request.Context(), c.Param("slug"), projectID, user.ID, id)
+	if err != nil {
+		h.importError(c, err)
+		return
+	}
+	c.JSON(http.StatusOK, im)
+}
+
+func (h *ImporterHandler) importError(c *gin.Context, err error) {
+	switch {
+	case errors.Is(err, service.ErrImportForbidden):
+		c.JSON(http.StatusForbidden, gin.H{"error": err.Error()})
+	case errors.Is(err, service.ErrImportWorkspace), errors.Is(err, service.ErrImportNotFound):
+		c.JSON(http.StatusNotFound, gin.H{"error": "Not found"})
+	case errors.Is(err, service.ErrImportBadFile), errors.Is(err, service.ErrImportNoName),
+		errors.Is(err, service.ErrImportEmpty):
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+	default:
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "Import failed"})
+	}
+}

apps/api/internal/handler/importer_test.go

@@ -0,0 +1,68 @@
+package handler_test
+
+import (
+	"encoding/json"
+	"net/http"
+	"testing"
+
+	"github.com/Devlaner/devlane/api/internal/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+// Bulk CSV import runs synchronously in tests (no RabbitMQ configured) and
+// creates one issue per row, mapping the state column by name. Covers #207.
+func TestImporter_CSV(t *testing.T) {
+	ts := testutil.NewTestServer(t)
+	w := testutil.SeedWorld(t, ts.DB)
+	state := testutil.CreateState(t, ts.DB, w.Project.ID, w.Workspace.ID)
+
+	base := "/api/workspaces/" + w.Workspace.Slug + "/projects/" + w.Project.ID.String() + "/importers/"
+	csv := "name,description,priority,state\n" +
+		"First task,Do the thing,high," + state.Name + "\n" +
+		"Second task,,low,\n" +
+		",skip me (no name),,\n" + // no name -> skipped
+		"Third task,More detail,bogus,Nonexistent State\n"
+
+	rr := ts.DoMultipart(http.MethodPost, base, "issues.csv", csv, w.Session)
+	require.Equal(t, http.StatusCreated, rr.Code, "body=%s", rr.Body.String())
+
+	var created map[string]any
+	require.NoError(t, json.Unmarshal(rr.Body.Bytes(), &created))
+	require.Equal(t, float64(3), created["total_count"], "3 rows have a name")
+	require.Equal(t, float64(3), created["processed_count"])
+	require.Equal(t, "completed", created["status"])
+
+	// It shows up in the list.
+	lr := ts.GET(base, w.Session)
+	require.Equal(t, http.StatusOK, lr.Code)
+	var list []map[string]any
+	require.NoError(t, json.Unmarshal(lr.Body.Bytes(), &list))
+	require.Len(t, list, 1)
+
+	// The three issues now exist on the project.
+	ir := ts.GET("/api/workspaces/"+w.Workspace.Slug+"/projects/"+w.Project.ID.String()+"/issues/", w.Session)
+	require.Equal(t, http.StatusOK, ir.Code)
+	var issues []map[string]any
+	require.NoError(t, json.Unmarshal(ir.Body.Bytes(), &issues))
+	require.Len(t, issues, 3)
+}
+
+// A CSV without a name/title/summary column is rejected.
+func TestImporter_RejectsNoNameColumn(t *testing.T) {
+	ts := testutil.NewTestServer(t)
+	w := testutil.SeedWorld(t, ts.DB)
+	base := "/api/workspaces/" + w.Workspace.Slug + "/projects/" + w.Project.ID.String() + "/importers/"
+	rr := ts.DoMultipart(http.MethodPost, base, "bad.csv", "foo,bar\n1,2\n", w.Session)
+	require.Equal(t, http.StatusBadRequest, rr.Code, "body=%s", rr.Body.String())
+}
+
+// A non-member cannot import into the project.
+func TestImporter_Forbidden(t *testing.T) {
+	ts := testutil.NewTestServer(t)
+	w := testutil.SeedWorld(t, ts.DB)
+	stranger := testutil.CreateUser(t, ts.DB)
+	session := testutil.LoginAs(t, ts.DB, stranger)
+	base := "/api/workspaces/" + w.Workspace.Slug + "/projects/" + w.Project.ID.String() + "/importers/"
+	rr := ts.DoMultipart(http.MethodPost, base, "x.csv", "name\nHello\n", session)
+	require.Equal(t, http.StatusForbidden, rr.Code)
+}

apps/api/internal/model/importer.go

@@ -0,0 +1,69 @@
+package model
+
+import (
+	"time"
+
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+)
+
+// Importer statuses.
+const (
+	ImportStatusQueued     = "queued"
+	ImportStatusProcessing = "processing"
+	ImportStatusCompleted  = "completed"
+	ImportStatusPartial    = "completed_with_errors"
+	ImportStatusFailed     = "failed"
+)
+
+// Importer services.
+const (
+	ImportServiceCSV = "csv"
+)
+
+// ImportRow is one parsed source record to be turned into an issue.
+type ImportRow struct {
+	Name        string `json:"name"`
+	Description string `json:"description,omitempty"`
+	Priority    string `json:"priority,omitempty"`
+	State       string `json:"state,omitempty"`
+}
+
+// ImporterData holds the parsed source rows, persisted in the data JSONB column
+// so the async worker can process them without re-reading the upload.
+type ImporterData struct {
+	Rows []ImportRow `json:"rows,omitempty"`
+}
+
+// Importer matches importers: one bulk-import job for a project.
+type Importer struct {
+	ID             uuid.UUID      `gorm:"type:uuid;primaryKey;default:gen_random_uuid()" json:"id"`
+	Service        string         `gorm:"type:varchar(50);not null" json:"service"`
+	Status         string         `gorm:"type:varchar(50);not null;default:queued" json:"status"`
+	Metadata       JSONMap        `gorm:"type:jsonb;default:'{}';serializer:json" json:"metadata,omitempty"`
+	Config         JSONMap        `gorm:"type:jsonb;default:'{}';serializer:json" json:"config,omitempty"`
+	Data           ImporterData   `gorm:"column:data;type:jsonb;serializer:json" json:"-"`
+	TotalCount     int            `gorm:"column:total_count;not null;default:0" json:"total_count"`
+	ProcessedCount int            `gorm:"column:processed_count;not null;default:0" json:"processed_count"`
+	ErrorCount     int            `gorm:"column:error_count;not null;default:0" json:"error_count"`
+	ErrorMessage   string         `gorm:"column:error_message;type:text" json:"error_message,omitempty"`
+	SourceFilename string         `gorm:"column:source_filename;type:varchar(512)" json:"source_filename,omitempty"`
+	ProjectID      *uuid.UUID     `gorm:"type:uuid" json:"project_id,omitempty"`
+	WorkspaceID    uuid.UUID      `gorm:"type:uuid;not null" json:"workspace_id"`
+	InitiatedByID  uuid.UUID      `gorm:"column:initiated_by_id;type:uuid;not null" json:"initiated_by_id"`
+	TokenID        *uuid.UUID     `gorm:"column:token_id;type:uuid" json:"-"`
+	CreatedAt      time.Time      `json:"created_at"`
+	UpdatedAt      time.Time      `json:"updated_at"`
+	DeletedAt      gorm.DeletedAt `gorm:"index" json:"-"`
+	CreatedByID    *uuid.UUID     `gorm:"type:uuid" json:"created_by_id,omitempty"`
+	UpdatedByID    *uuid.UUID     `gorm:"type:uuid" json:"updated_by_id,omitempty"`
+}
+
+func (Importer) TableName() string { return "importers" }
+
+func (i *Importer) BeforeCreate(tx *gorm.DB) error {
+	if i.ID == uuid.Nil {
+		i.ID = uuid.New()
+	}
+	return nil
+}

apps/api/internal/queue/consumer.go

@@ -160,6 +160,23 @@ func HandleWebhook(deliverer func(ctx context.Context, p WebhookPayload) error)
 	}
 }
 
+// HandleImport parses an import_run task and runs the given importer.
+func HandleImport(runner func(ctx context.Context, importerID string) error) TaskHandler {
+	return func(ctx context.Context, queue string, body []byte) error {
+		var msg struct {
+			Type    string        `json:"type"`
+			Payload ImportPayload `json:"payload"`
+		}
+		if err := json.Unmarshal(body, &msg); err != nil {
+			return err
+		}
+		if msg.Type != TaskImportRun {
+			return nil
+		}
+		return runner(ctx, msg.Payload.ImporterID)
+	}
+}
+
 // NoopEmailSender is a no-op sender (log only). Replace with real SMTP in production.
 func NoopEmailSender(log *slog.Logger) func(ctx context.Context, to, subject, body string) error {
 	return func(ctx context.Context, to, subject, body string) error {

apps/api/internal/queue/queue.go

@@ -14,13 +14,15 @@ import (
 const (
 	QueueEmails   = "devlane.emails"
 	QueueWebhooks = "devlane.webhooks"
+	QueueImports  = "devlane.imports"
 	QueueDefault  = "devlane.default"
 )
 
 // Task types for routing or payload identification.
 const (
 	TaskSendEmail      = "send_email"
 	TaskWebhookDeliver = "webhook_deliver"
+	TaskImportRun      = "import_run"
 )
 
 // SendEmailPayload is the payload for send_email task.
@@ -43,6 +45,12 @@ type WebhookPayload struct {
 	Payload     map[string]interface{} `json:"payload"`
 }
 
+// ImportPayload is the payload for an import_run task. The rows themselves live
+// on the importer row (data JSONB); only the id is carried through the queue.
+type ImportPayload struct {
+	ImporterID string `json:"importer_id"`
+}
+
 // Publisher publishes tasks to RabbitMQ.
 type Publisher struct {
 	ch     *amqp.Channel
@@ -52,13 +60,13 @@ type Publisher struct {
 
 // NewPublisher declares queues and returns a publisher.
 func NewPublisher(ch *amqp.Channel, log *slog.Logger) (*Publisher, error) {
-	for _, q := range []string{QueueEmails, QueueWebhooks, QueueDefault} {
+	for _, q := range []string{QueueEmails, QueueWebhooks, QueueImports, QueueDefault} {
 		if _, err := ch.QueueDeclare(q, true, false, false, false, nil); err != nil {
 			return nil, fmt.Errorf("declare queue %s: %w", q, err)
 		}
 	}
 	return &Publisher{ch: ch, log: log, queues: map[string]bool{
-		QueueEmails: true, QueueWebhooks: true, QueueDefault: true,
+		QueueEmails: true, QueueWebhooks: true, QueueImports: true, QueueDefault: true,
 	}}, nil
 }
 
@@ -100,3 +108,14 @@ func (p *Publisher) PublishWebhook(ctx context.Context, payload WebhookPayload)
 		"payload": payload,
 	})
 }
+
+// PublishImport enqueues an import_run task.
+func (p *Publisher) PublishImport(ctx context.Context, payload ImportPayload) error {
+	if p.log != nil {
+		p.log.Debug("queue publish import", "importer_id", payload.ImporterID)
+	}
+	return p.PublishJSON(ctx, QueueImports, map[string]interface{}{
+		"type":    TaskImportRun,
+		"payload": payload,
+	})
+}