diff --git a/apps/api/cmd/api/main.go b/apps/api/cmd/api/main.go index 9f95c315..5ea6d5a0 100644 --- a/apps/api/cmd/api/main.go +++ b/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) } } diff --git a/apps/api/internal/handler/importer.go b/apps/api/internal/handler/importer.go new file mode 100644 index 00000000..0856914d --- /dev/null +++ b/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"}) + } +} diff --git a/apps/api/internal/handler/importer_test.go b/apps/api/internal/handler/importer_test.go new file mode 100644 index 00000000..2cdde218 --- /dev/null +++ b/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) +} diff --git a/apps/api/internal/model/importer.go b/apps/api/internal/model/importer.go new file mode 100644 index 00000000..87f6ae06 --- /dev/null +++ b/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 +} diff --git a/apps/api/internal/queue/consumer.go b/apps/api/internal/queue/consumer.go index c0e36c48..3aadca01 100644 --- a/apps/api/internal/queue/consumer.go +++ b/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 { diff --git a/apps/api/internal/queue/queue.go b/apps/api/internal/queue/queue.go index ce1a2e34..5395df4b 100644 --- a/apps/api/internal/queue/queue.go +++ b/apps/api/internal/queue/queue.go @@ -14,6 +14,7 @@ import ( const ( QueueEmails = "devlane.emails" QueueWebhooks = "devlane.webhooks" + QueueImports = "devlane.imports" QueueDefault = "devlane.default" ) @@ -21,6 +22,7 @@ const ( 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, + }) +} diff --git a/apps/api/internal/router/router.go b/apps/api/internal/router/router.go index 52e3d22a..39e9da98 100644 --- a/apps/api/internal/router/router.go +++ b/apps/api/internal/router/router.go @@ -35,7 +35,9 @@ type Config struct { } // New builds and returns the Gin engine with /api/ and /auth/ routes. -func New(cfg Config) *gin.Engine { +// New builds the Gin engine and also returns the ImporterService so the caller +// (cmd/api) can register the background import worker on the task queue. +func New(cfg Config) (*gin.Engine, *service.ImporterService) { if cfg.Log == nil { cfg.Log = slog.Default() } @@ -180,6 +182,8 @@ func New(cfg Config) *gin.Engine { issueSvc.SetReactionStore(issueReactionStore) issueSvc.SetStateStore(stateStore) issueSvc.SetLabelStore(labelStore) + importerStore := store.NewImporterStore(cfg.DB) + importerSvc := service.NewImporterService(importerStore, workspaceStore, projectStore, stateStore, issueSvc, cfg.Queue, cfg.Log) commentReactionStore := store.NewCommentReactionStore(cfg.DB) commentSvc := service.NewCommentService(commentStore, issueStore, projectStore, workspaceStore) commentSvc.SetReactionStore(commentReactionStore) @@ -251,6 +255,7 @@ func New(cfg Config) *gin.Engine { intakeHandler := &handler.IntakeHandler{Intake: intakeSvc} webhookHandler := &handler.WebhookHandler{Webhooks: webhookSvc} issueHandler := &handler.IssueHandler{Issue: issueSvc} + importerHandler := &handler.ImporterHandler{Importers: importerSvc} issueLinkHandler := &handler.IssueLinkHandler{Issue: issueSvc} attachmentHandler := &handler.AttachmentHandler{Attachment: attachmentSvc} epicHandler := &handler.EpicHandler{Issue: issueSvc} @@ -379,6 +384,11 @@ func New(cfg Config) *gin.Engine { api.GET("/workspaces/:slug/projects/:projectId/intake-issues/count/", intakeHandler.Count) api.PATCH("/workspaces/:slug/projects/:projectId/intake-issues/:pk/", intakeHandler.Transition) + // Bulk import (CSV) for a project. + api.GET("/workspaces/:slug/projects/:projectId/importers/", importerHandler.List) + api.POST("/workspaces/:slug/projects/:projectId/importers/", importerHandler.Create) + api.GET("/workspaces/:slug/projects/:projectId/importers/:importerId/", importerHandler.Get) + api.GET("/workspaces/:slug/projects/:projectId/issues/", issueHandler.List) api.POST("/workspaces/:slug/projects/:projectId/issues/", issueHandler.Create) api.GET("/workspaces/:slug/projects/:projectId/issues/:pk/", issueHandler.Get) @@ -605,5 +615,5 @@ func New(cfg Config) *gin.Engine { }) } - return r + return r, importerSvc } diff --git a/apps/api/internal/service/importer.go b/apps/api/internal/service/importer.go new file mode 100644 index 00000000..4835d4fc --- /dev/null +++ b/apps/api/internal/service/importer.go @@ -0,0 +1,330 @@ +package service + +import ( + "context" + "encoding/csv" + "errors" + "fmt" + "io" + "log/slog" + "strings" + + "github.com/Devlaner/devlane/api/internal/model" + "github.com/Devlaner/devlane/api/internal/queue" + "github.com/Devlaner/devlane/api/internal/store" + "github.com/google/uuid" +) + +var ( + ErrImportWorkspace = errors.New("workspace not found") + ErrImportForbidden = errors.New("you do not have access to this project") + ErrImportNotFound = errors.New("import not found") + ErrImportBadFile = errors.New("could not parse the CSV file") + ErrImportNoName = errors.New("the CSV must have a name or title column") + ErrImportEmpty = errors.New("the CSV has no data rows") +) + +// maxImportRows caps a single import so a huge upload cannot exhaust memory or +// the queue payload. Anything above is rejected with a clear error. +const maxImportRows = 5000 + +// ImporterService parses uploaded files into issues for a project. CSV is the +// only supported source today; Jira/GitHub bulk import are planned follow-ups. +type ImporterService struct { + importers *store.ImporterStore + ws *store.WorkspaceStore + ps *store.ProjectStore + states *store.StateStore + issues *IssueService + queue *queue.Publisher // optional; nil -> run synchronously + log *slog.Logger +} + +func NewImporterService( + importers *store.ImporterStore, + ws *store.WorkspaceStore, + ps *store.ProjectStore, + states *store.StateStore, + issues *IssueService, + q *queue.Publisher, + log *slog.Logger, +) *ImporterService { + return &ImporterService{importers: importers, ws: ws, ps: ps, states: states, issues: issues, queue: q, log: log} +} + +// ensureAccess mirrors IssueService.ensureProjectAccess: caller must be a member +// of the workspace and the project must belong to it. +func (s *ImporterService) ensureAccess(ctx context.Context, slug string, projectID, userID uuid.UUID) (*model.Workspace, error) { + wrk, err := s.ws.GetBySlug(ctx, slug) + if err != nil { + return nil, ErrImportWorkspace + } + ok, _ := s.ws.IsMember(ctx, wrk.ID, userID) + if !ok { + return nil, ErrImportForbidden + } + in, _ := s.ps.IsInWorkspace(ctx, projectID, wrk.ID) + if !in { + return nil, ErrImportNotFound + } + return wrk, nil +} + +// CreateCSV parses a CSV upload into pending import rows, persists the job, and +// enqueues it (or runs it inline when no queue is configured). +func (s *ImporterService) CreateCSV(ctx context.Context, slug string, projectID, userID uuid.UUID, filename string, r io.Reader) (*model.Importer, error) { + wrk, err := s.ensureAccess(ctx, slug, projectID, userID) + if err != nil { + return nil, err + } + + rows, err := parseCSV(r) + if err != nil { + return nil, err + } + if len(rows) == 0 { + return nil, ErrImportEmpty + } + if len(rows) > maxImportRows { + return nil, fmt.Errorf("%w: at most %d rows are supported per import", ErrImportBadFile, maxImportRows) + } + + pid := projectID + im := &model.Importer{ + Service: model.ImportServiceCSV, + Status: model.ImportStatusQueued, + Data: model.ImporterData{Rows: rows}, + TotalCount: len(rows), + SourceFilename: filename, + ProjectID: &pid, + WorkspaceID: wrk.ID, + InitiatedByID: userID, + CreatedByID: &userID, + UpdatedByID: &userID, + } + if err := s.importers.Create(ctx, im); err != nil { + return nil, err + } + + ranInline := false + if s.queue != nil { + if err := s.queue.PublishImport(ctx, queue.ImportPayload{ImporterID: im.ID.String()}); err != nil { + if s.log != nil { + s.log.Warn("import enqueue failed, running inline", "importer_id", im.ID, "error", err) + } + _ = s.Run(ctx, im.ID.String()) + ranInline = true + } + } else { + // No queue available (optional infra): process synchronously so the + // feature still works. + _ = s.Run(ctx, im.ID.String()) + ranInline = true + } + // When processed inline, return the finished job so the caller sees the real + // status/counts instead of the initial "queued" snapshot. + if ranInline { + if fresh, _ := s.importers.Get(ctx, im.ID); fresh != nil { + return fresh, nil + } + } + return im, nil +} + +// List returns a project's imports, newest first. +func (s *ImporterService) List(ctx context.Context, slug string, projectID, userID uuid.UUID) ([]model.Importer, error) { + if _, err := s.ensureAccess(ctx, slug, projectID, userID); err != nil { + return nil, err + } + return s.importers.ListByProject(ctx, projectID) +} + +// Get returns a single import (for status polling). +func (s *ImporterService) Get(ctx context.Context, slug string, projectID, userID, id uuid.UUID) (*model.Importer, error) { + if _, err := s.ensureAccess(ctx, slug, projectID, userID); err != nil { + return nil, err + } + im, err := s.importers.GetByID(ctx, projectID, id) + if err != nil { + return nil, err + } + if im == nil { + return nil, ErrImportNotFound + } + return im, nil +} + +// Run processes an import job: create one issue per parsed row, tracking +// progress and per-row failures. Invoked by the queue worker (or inline). +func (s *ImporterService) Run(ctx context.Context, importerID string) error { + id, err := uuid.Parse(importerID) + if err != nil { + return err + } + im, err := s.importers.Get(ctx, id) + if err != nil { + return err + } + if im == nil { + // The importer row is gone (deleted, or a stale queue message). Nothing + // to process; log so it isn't silently swallowed, but ack the message. + if s.log != nil { + s.log.Warn("import run: importer not found", "importer_id", importerID) + } + return nil + } + // Skip anything already finished OR in-flight: on a redelivered message an + // import left in "processing" (e.g. a mid-run crash) must not be replayed + // from row zero, which would create duplicate issues. + if im.Status == model.ImportStatusCompleted || + im.Status == model.ImportStatusPartial || + im.Status == model.ImportStatusProcessing { + return nil + } + if im.ProjectID == nil { + im.Status = model.ImportStatusFailed + im.ErrorMessage = "import has no target project" + return s.importers.UpdateProgress(ctx, im) + } + + wrk, err := s.ws.GetByID(ctx, im.WorkspaceID) + if err != nil || wrk == nil { + im.Status = model.ImportStatusFailed + im.ErrorMessage = "workspace not found" + return s.importers.UpdateProgress(ctx, im) + } + + // Map state names -> ids once for this project. + stateByName := map[string]uuid.UUID{} + if states, err := s.states.ListByProjectID(ctx, *im.ProjectID); err == nil { + for i := range states { + stateByName[strings.ToLower(strings.TrimSpace(states[i].Name))] = states[i].ID + } + } + + im.Status = model.ImportStatusProcessing + im.ProcessedCount = 0 + im.ErrorCount = 0 + im.ErrorMessage = "" + _ = s.importers.UpdateProgress(ctx, im) + + var firstErr string + for _, row := range im.Data.Rows { + var stateID *uuid.UUID + if row.State != "" { + if sid, ok := stateByName[strings.ToLower(strings.TrimSpace(row.State))]; ok { + sid := sid + stateID = &sid + } + } + _, cerr := s.issues.Create(ctx, wrk.Slug, *im.ProjectID, im.InitiatedByID, + row.Name, row.Description, normalizePriority(row.Priority), + stateID, nil, nil, nil, nil, nil, false) + if cerr != nil { + im.ErrorCount++ + if firstErr == "" { + firstErr = cerr.Error() + } + } else { + im.ProcessedCount++ + } + _ = s.importers.UpdateProgress(ctx, im) + } + + switch { + case im.ProcessedCount == 0: + im.Status = model.ImportStatusFailed + case im.ErrorCount > 0: + im.Status = model.ImportStatusPartial + default: + im.Status = model.ImportStatusCompleted + } + if firstErr != "" { + im.ErrorMessage = fmt.Sprintf("%d row(s) failed; first error: %s", im.ErrorCount, firstErr) + } + return s.importers.UpdateProgress(ctx, im) +} + +// parseCSV reads a CSV with a header row and maps common columns (name/title, +// description, priority, state/status) into import rows. Rows without a name +// are skipped. +func parseCSV(r io.Reader) ([]model.ImportRow, error) { + cr := csv.NewReader(r) + cr.TrimLeadingSpace = true + cr.FieldsPerRecord = -1 // tolerate ragged rows + + header, err := cr.Read() + if err != nil { + if errors.Is(err, io.EOF) { + return nil, ErrImportEmpty + } + return nil, fmt.Errorf("%w: %v", ErrImportBadFile, err) + } + + col := map[string]int{} + for i, h := range header { + col[strings.ToLower(strings.TrimSpace(h))] = i + } + nameIdx, ok := firstIndex(col, "name", "title", "summary") + if !ok { + return nil, ErrImportNoName + } + descIdx, _ := firstIndex(col, "description", "body", "details") + prioIdx, _ := firstIndex(col, "priority") + stateIdx, _ := firstIndex(col, "state", "status") + + var rows []model.ImportRow + for { + rec, err := cr.Read() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrImportBadFile, err) + } + name := strings.TrimSpace(field(rec, nameIdx)) + if name == "" { + continue // skip rows with no title + } + rows = append(rows, model.ImportRow{ + Name: name, + Description: strings.TrimSpace(field(rec, descIdx)), + Priority: strings.TrimSpace(field(rec, prioIdx)), + State: strings.TrimSpace(field(rec, stateIdx)), + }) + } + return rows, nil +} + +func firstIndex(col map[string]int, names ...string) (int, bool) { + for _, n := range names { + if i, ok := col[n]; ok { + return i, true + } + } + return -1, false +} + +func field(rec []string, idx int) string { + if idx < 0 || idx >= len(rec) { + return "" + } + return rec[idx] +} + +// normalizePriority maps free-text priority to the app's allowed set, defaulting +// to "none" for anything unrecognized. +func normalizePriority(p string) string { + switch strings.ToLower(strings.TrimSpace(p)) { + case "urgent": + return "urgent" + case "high": + return "high" + case "medium", "med": + return "medium" + case "low": + return "low" + default: + return "none" + } +} diff --git a/apps/api/internal/service/importer_internal_test.go b/apps/api/internal/service/importer_internal_test.go new file mode 100644 index 00000000..e8340eda --- /dev/null +++ b/apps/api/internal/service/importer_internal_test.go @@ -0,0 +1,51 @@ +package service + +import ( + "strings" + "testing" +) + +func TestParseCSV_MapsColumnsAndSkipsEmptyNames(t *testing.T) { + in := "Title,Description,Priority,Status\n" + + "Fix login,Users cannot sign in,High,In Progress\n" + + " ,orphan row with no title,,\n" + + "Add search,,,Todo\n" + rows, err := parseCSV(strings.NewReader(in)) + if err != nil { + t.Fatalf("parseCSV: %v", err) + } + if len(rows) != 2 { + t.Fatalf("expected 2 rows (empty-name skipped), got %d", len(rows)) + } + if rows[0].Name != "Fix login" || rows[0].Description != "Users cannot sign in" || + rows[0].Priority != "High" || rows[0].State != "In Progress" { + t.Errorf("row0 mapped wrong: %+v", rows[0]) + } + if rows[1].Name != "Add search" || rows[1].State != "Todo" { + t.Errorf("row1 mapped wrong: %+v", rows[1]) + } +} + +func TestParseCSV_RequiresNameColumn(t *testing.T) { + if _, err := parseCSV(strings.NewReader("foo,bar\n1,2\n")); err != ErrImportNoName { + t.Errorf("expected ErrImportNoName, got %v", err) + } +} + +func TestParseCSV_EmptyIsRejected(t *testing.T) { + if _, err := parseCSV(strings.NewReader("")); err != ErrImportEmpty { + t.Errorf("expected ErrImportEmpty, got %v", err) + } +} + +func TestNormalizePriority(t *testing.T) { + cases := map[string]string{ + "Urgent": "urgent", "HIGH": "high", "med": "medium", "medium": "medium", + "low": "low", "": "none", "whatever": "none", + } + for in, want := range cases { + if got := normalizePriority(in); got != want { + t.Errorf("normalizePriority(%q) = %q, want %q", in, got, want) + } + } +} diff --git a/apps/api/internal/store/importer.go b/apps/api/internal/store/importer.go new file mode 100644 index 00000000..e75ce5f4 --- /dev/null +++ b/apps/api/internal/store/importer.go @@ -0,0 +1,75 @@ +package store + +import ( + "context" + "errors" + + "github.com/Devlaner/devlane/api/internal/model" + "github.com/google/uuid" + "gorm.io/gorm" +) + +// ImporterStore is the data-access layer for bulk-import jobs. +type ImporterStore struct { + db *gorm.DB +} + +func NewImporterStore(db *gorm.DB) *ImporterStore { + return &ImporterStore{db: db} +} + +func (s *ImporterStore) Create(ctx context.Context, im *model.Importer) error { + return s.db.WithContext(ctx).Create(im).Error +} + +// GetByID returns the importer scoped to a project, or (nil, nil) if not found. +func (s *ImporterStore) GetByID(ctx context.Context, projectID, id uuid.UUID) (*model.Importer, error) { + var im model.Importer + err := s.db.WithContext(ctx). + Where("id = ? AND project_id = ?", id, projectID). + First(&im).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &im, nil +} + +// Get returns the importer by id alone (used by the worker), or (nil, nil). +func (s *ImporterStore) Get(ctx context.Context, id uuid.UUID) (*model.Importer, error) { + var im model.Importer + err := s.db.WithContext(ctx).Where("id = ?", id).First(&im).Error + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return &im, nil +} + +// ListByProject returns a project's imports, newest first. Row data is not +// selected, keeping list responses small. +func (s *ImporterStore) ListByProject(ctx context.Context, projectID uuid.UUID) ([]model.Importer, error) { + var list []model.Importer + err := s.db.WithContext(ctx). + Omit("data"). + Where("project_id = ?", projectID). + Order("created_at DESC"). + Find(&list).Error + return list, err +} + +// UpdateProgress persists the current status and counters. +func (s *ImporterStore) UpdateProgress(ctx context.Context, im *model.Importer) error { + return s.db.WithContext(ctx).Model(im). + Select("status", "processed_count", "error_count", "error_message", "updated_at"). + Updates(map[string]interface{}{ + "status": im.Status, + "processed_count": im.ProcessedCount, + "error_count": im.ErrorCount, + "error_message": im.ErrorMessage, + }).Error +} diff --git a/apps/api/internal/testutil/http.go b/apps/api/internal/testutil/http.go index 757534a8..4c07d7bf 100644 --- a/apps/api/internal/testutil/http.go +++ b/apps/api/internal/testutil/http.go @@ -4,6 +4,7 @@ import ( "bytes" "encoding/json" "io" + "mime/multipart" "net/http" "net/http/httptest" "strings" @@ -12,6 +13,32 @@ import ( "github.com/Devlaner/devlane/api/internal/middleware" ) +// DoMultipart posts a single-file multipart/form-data request (field "file") +// through the router, for endpoints that accept uploads. +func (ts *TestServer) DoMultipart(method, path, filename, content, sessionKey string) *httptest.ResponseRecorder { + ts.T.Helper() + var buf bytes.Buffer + w := multipart.NewWriter(&buf) + fw, err := w.CreateFormFile("file", filename) + if err != nil { + ts.T.Fatalf("create form file: %v", err) + } + if _, err := fw.Write([]byte(content)); err != nil { + ts.T.Fatalf("write form file: %v", err) + } + if err := w.Close(); err != nil { + ts.T.Fatalf("close multipart: %v", err) + } + req := httptest.NewRequest(method, path, &buf) + req.Header.Set("Content-Type", w.FormDataContentType()) + if sessionKey != "" { + req.AddCookie(&http.Cookie{Name: middleware.SessionCookieName, Value: sessionKey}) + } + rr := httptest.NewRecorder() + ts.Router.ServeHTTP(rr, req) + return rr +} + // Do dispatches an HTTP request through the gin router and returns the recorder. // body may be nil (no body), a string, []byte, io.Reader, or any JSON-marshallable value. // sessionKey, if non-empty, is attached as the session_id cookie. diff --git a/apps/api/internal/testutil/router.go b/apps/api/internal/testutil/router.go index c49bb4ea..18232a3e 100644 --- a/apps/api/internal/testutil/router.go +++ b/apps/api/internal/testutil/router.go @@ -40,10 +40,11 @@ func NewTestServer(t testing.TB) *TestServer { AppBaseURL: "http://localhost:5173", APIPublicURL: "http://localhost:8080", } + eng, _ := router.New(cfg) return &TestServer{ T: t, DB: db, - Router: router.New(cfg), + Router: eng, } } diff --git a/apps/api/migrations/000012_importer_progress.down.sql b/apps/api/migrations/000012_importer_progress.down.sql new file mode 100644 index 00000000..ef1aee65 --- /dev/null +++ b/apps/api/migrations/000012_importer_progress.down.sql @@ -0,0 +1,12 @@ +DROP INDEX IF EXISTS idx_importers_project; + +ALTER TABLE importers + DROP COLUMN IF EXISTS total_count, + DROP COLUMN IF EXISTS processed_count, + DROP COLUMN IF EXISTS error_count, + DROP COLUMN IF EXISTS error_message, + DROP COLUMN IF EXISTS source_filename; + +-- Restore the NOT NULL constraint on token_id. This only succeeds if no rows +-- with a null token_id remain (true on a clean rollback). +ALTER TABLE importers ALTER COLUMN token_id SET NOT NULL; diff --git a/apps/api/migrations/000012_importer_progress.up.sql b/apps/api/migrations/000012_importer_progress.up.sql new file mode 100644 index 00000000..83672f70 --- /dev/null +++ b/apps/api/migrations/000012_importer_progress.up.sql @@ -0,0 +1,15 @@ +-- Adapt the pre-existing (unused) importers table for user-initiated bulk +-- imports. The table was scaffolded around a token-based flow (token_id NOT +-- NULL referencing api_tokens); a CSV import is initiated by a signed-in user +-- through the UI and has no API token, so relax that constraint and add the +-- progress/columns the importer service tracks. +ALTER TABLE importers ALTER COLUMN token_id DROP NOT NULL; + +ALTER TABLE importers + ADD COLUMN IF NOT EXISTS total_count INTEGER NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS processed_count INTEGER NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS error_count INTEGER NOT NULL DEFAULT 0, + ADD COLUMN IF NOT EXISTS error_message TEXT, + ADD COLUMN IF NOT EXISTS source_filename VARCHAR(512); + +CREATE INDEX IF NOT EXISTS idx_importers_project ON importers (project_id, created_at DESC); diff --git a/apps/web/src/api/types.ts b/apps/web/src/api/types.ts index 061c73f5..ddaae18c 100644 --- a/apps/web/src/api/types.ts +++ b/apps/web/src/api/types.ts @@ -356,6 +356,22 @@ export interface FavoriteApiResponse { updated_at?: string; } +/** A bulk-import job (e.g. CSV) for a project. */ +export interface ImporterApiResponse { + id: string; + service: string; + status: 'queued' | 'processing' | 'completed' | 'completed_with_errors' | 'failed'; + total_count: number; + processed_count: number; + error_count: number; + error_message?: string; + source_filename?: string; + project_id?: string; + workspace_id: string; + created_at?: string; + updated_at?: string; +} + /** An outbound workspace webhook. */ export interface WebhookApiResponse { id: string; diff --git a/apps/web/src/components/work-item/ImportCSVModal.tsx b/apps/web/src/components/work-item/ImportCSVModal.tsx new file mode 100644 index 00000000..e786285a --- /dev/null +++ b/apps/web/src/components/work-item/ImportCSVModal.tsx @@ -0,0 +1,175 @@ +import { useEffect, useRef, useState } from 'react'; +import { Modal, Button } from '../ui'; +import { importerService } from '../../services/importerService'; +import type { ImporterApiResponse } from '../../api/types'; + +interface ImportCSVModalProps { + open: boolean; + onClose: () => void; + workspaceSlug: string; + projectId: string; + /** Called once an import finishes so the caller can refresh the issue list. */ + onImported?: () => void; +} + +const isTerminal = (s: ImporterApiResponse['status']) => + s === 'completed' || s === 'completed_with_errors' || s === 'failed'; + +/** + * Upload a CSV of work items and watch the import progress (issue #207). The + * CSV needs a name/title/summary column; description, priority, and state/status + * columns are mapped when present. Progress is polled until the job finishes. + */ +export function ImportCSVModal({ + open, + onClose, + workspaceSlug, + projectId, + onImported, +}: ImportCSVModalProps) { + const [file, setFile] = useState(null); + const [job, setJob] = useState(null); + const [uploading, setUploading] = useState(false); + const [error, setError] = useState(null); + const notifiedRef = useRef(false); + // Held in a ref so the poll effect below doesn't depend on onImported's + // identity — the parent passes a fresh function each render, which would + // otherwise tear down and restart the 1s poll timer on every re-render. + const onImportedRef = useRef(onImported); + onImportedRef.current = onImported; + + // Reset when the modal is (re)opened. + useEffect(() => { + if (open) { + setFile(null); + setJob(null); + setUploading(false); + setError(null); + notifiedRef.current = false; + } + }, [open]); + + // Poll until the job reaches a terminal state. + useEffect(() => { + if (!job || isTerminal(job.status)) { + if (job && isTerminal(job.status) && !notifiedRef.current) { + notifiedRef.current = true; + onImportedRef.current?.(); + } + return; + } + let cancelled = false; + const timer = setTimeout(async () => { + try { + const next = await importerService.get(workspaceSlug, projectId, job.id); + if (!cancelled) setJob(next); + } catch { + // Keep the last known state; a transient poll error shouldn't crash the UI. + } + }, 1000); + return () => { + cancelled = true; + clearTimeout(timer); + }; + }, [job, workspaceSlug, projectId]); + + const handleUpload = async () => { + if (!file) return; + setUploading(true); + setError(null); + try { + const created = await importerService.createCSV(workspaceSlug, projectId, file); + setJob(created); + } catch (err: unknown) { + const res = (err as { response?: { status?: number; data?: { error?: string } } })?.response; + setError( + res?.data?.error || + (res?.status === 403 + ? 'You do not have access to import into this project.' + : 'Could not start the import. Check the file and try again.'), + ); + } finally { + setUploading(false); + } + }; + + const percent = + job && job.total_count > 0 + ? Math.round(((job.processed_count + job.error_count) / job.total_count) * 100) + : 0; + + return ( + + {job ? ( +
+
+
+ + {job.status === 'processing' || job.status === 'queued' + ? 'Importing…' + : job.status === 'failed' + ? 'Import failed' + : job.error_count > 0 + ? 'Imported with some errors' + : 'Import complete'} + + {percent}% +
+
+
+
+
+
+
+
Total
+
{job.total_count}
+
+
+
Created
+
{job.processed_count}
+
+
+
Errors
+
{job.error_count}
+
+
+ {job.error_message && ( +

{job.error_message}

+ )} + {isTerminal(job.status) && ( +
+ +
+ )} +
+ ) : ( +
+

+ Upload a CSV with a name (or{' '} + title) column. Optional{' '} + description, priority, + and state columns are mapped when present. +

+ setFile(e.target.files?.[0] ?? null)} + className="block w-full text-sm text-(--txt-secondary) file:mr-3 file:rounded-(--radius-md) file:border file:border-(--border-subtle) file:bg-(--bg-surface-1) file:px-3 file:py-1.5 file:text-sm file:text-(--txt-primary) hover:file:bg-(--bg-layer-1-hover)" + /> + {error &&

{error}

} +
+ + +
+
+ )} + + ); +} diff --git a/apps/web/src/pages/IssueListPage.tsx b/apps/web/src/pages/IssueListPage.tsx index fa059716..0530a053 100644 --- a/apps/web/src/pages/IssueListPage.tsx +++ b/apps/web/src/pages/IssueListPage.tsx @@ -2,6 +2,7 @@ import { useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react'; import { useParams, useSearchParams } from 'react-router-dom'; import { Button } from '../components/ui'; import { CreateWorkItemModal } from '../components/CreateWorkItemModal'; +import { ImportCSVModal } from '../components/work-item/ImportCSVModal'; import { workspaceService } from '../services/workspaceService'; import { projectService } from '../services/projectService'; import { issueService } from '../services/issueService'; @@ -97,6 +98,7 @@ export function IssueListPage() { const [project, setProject] = useState(null); const [projects, setProjects] = useState([]); const [issues, setIssues] = useState([]); + const [importOpen, setImportOpen] = useState(false); const [states, setStates] = useState([]); const [labels, setLabels] = useState([]); const [cycles, setCycles] = useState([]); @@ -683,6 +685,9 @@ export function IssueListPage() { +
{issues.length === 0 ? ( @@ -831,6 +836,13 @@ export function IssueListPage() { onSave={handleCreateSave} createError={createError} /> + setImportOpen(false)} + workspaceSlug={workspace.slug} + projectId={project.id} + onImported={refetchIssues} + /> ); } diff --git a/apps/web/src/services/importerService.ts b/apps/web/src/services/importerService.ts new file mode 100644 index 00000000..ffa01bcb --- /dev/null +++ b/apps/web/src/services/importerService.ts @@ -0,0 +1,40 @@ +import { apiClient } from '../api/client'; +import type { ImporterApiResponse } from '../api/types'; + +const base = (workspaceSlug: string, projectId: string) => + `/api/workspaces/${encodeURIComponent(workspaceSlug)}/projects/${encodeURIComponent(projectId)}/importers/`; + +/** Bulk import (CSV) jobs for a project. */ +export const importerService = { + /** Upload a CSV file to start an import. */ + async createCSV( + workspaceSlug: string, + projectId: string, + file: File, + ): Promise { + const form = new FormData(); + form.append('file', file); + // apiClient strips Content-Type for FormData so the browser sets the boundary. + const { data } = await apiClient.post( + base(workspaceSlug, projectId), + form, + ); + return data; + }, + + async list(workspaceSlug: string, projectId: string): Promise { + const { data } = await apiClient.get(base(workspaceSlug, projectId)); + return Array.isArray(data) ? data : []; + }, + + async get( + workspaceSlug: string, + projectId: string, + importerId: string, + ): Promise { + const { data } = await apiClient.get( + `${base(workspaceSlug, projectId)}${encodeURIComponent(importerId)}/`, + ); + return data; + }, +};