diff --git a/PROGRESS.md b/PROGRESS.md index 52e0db9..d4da1c2 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -6,3 +6,4 @@ - Phase 3 (auth): argon2id (PHC, spec params), login/register rate limiting, opaque Mongo sessions (30d sliding, TTL index), CSRF double-submit, bootstrap admin with forced password change, RBAC middleware + matrix tests, OIDC code flow with PKCE + email linking (verified-only), full integration suite incl. fake IdP — all green. - Phase 4 (base UI shell): embedded templates + static assets (no build step), §10 beige/dark tokens with radius 2px, theme toggle persisted to localStorage + profile, login/register/change-password pages, profile page (avatar upload w/ MIME check, bio, contacts, arbitrary extra fields, version-conflict 409), role-based navigation, /files/{id} serving (session or signed token), CSP without inline scripts + security headers — unit + integration tests green, stack verified live. - Phase 5 (admin): customer CRUD wizard with per-system encrypted credentials + test connection (jira/ado/youtrack/demo connectors), user management (roles, disable w/ session revoke, force-reset, delete, self-lockout guards), runtime settings doc, audit log on every privileged mutation + list API, service-status panel (mongo/atomizer/WP health + latency, late-bound breaker/sync/jobs providers), tabbed admin UI - tests green. Fixed mongod WT_PANIC: container nofile ulimit was 1024, raised to 64000 in compose. +- Phase 6 (sync workers): per-customer pollers with reconcile loop (start/stop/interval changes), §5.3 idempotent upsert keyed (system,key,customerId) w/ content-hash change detection, upstream edits refresh only while imported (timeline+notification after), attachment caching to GridFS, orphan flagging (never deletes), admin sync-now trigger + worker statuses, default budget prefill — demo-type integration tests green. diff --git a/cmd/app/main.go b/cmd/app/main.go index caea867..a87c549 100644 --- a/cmd/app/main.go +++ b/cmd/app/main.go @@ -3,6 +3,7 @@ package main import ( "context" + "encoding/json" "fmt" "log/slog" "os" @@ -12,10 +13,13 @@ import ( "bountyboard/internal/auth" "bountyboard/internal/config" + "bountyboard/internal/crypto" + "bountyboard/internal/domain" "bountyboard/internal/files" httpx "bountyboard/internal/http" "bountyboard/internal/metrics" "bountyboard/internal/store" + syncpkg "bountyboard/internal/sync" ) func main() { @@ -57,12 +61,33 @@ func run() error { } reg := metrics.NewRegistry() - srv := httpx.New(cfg, log, reg, st, files.NewStore(st.DB, cfg.MaxUploadMB)) + fileStore := files.NewStore(st.DB, cfg.MaxUploadMB) + srv := httpx.New(cfg, log, reg, st, fileStore) srv.AddReadinessCheck(httpx.ReadinessCheck{ Name: "mongo", Required: true, Probe: st.Ping, }) + syncMgr := syncpkg.NewManager(st, fileStore, log, reg, + func(c *domain.Customer) (syncpkg.Credentials, error) { + if c.Ticketing.CredentialsEnc == "" { + return syncpkg.Credentials{}, nil + } + plain, err := crypto.Open(cfg.CredentialsEncKey, c.Ticketing.CredentialsEnc) + if err != nil { + return nil, err + } + var creds syncpkg.Credentials + if err := json.Unmarshal(plain, &creds); err != nil { + return nil, err + } + return creds, nil + }, + srv.Publish) + srv.SetSyncStatusProvider(syncMgr) + srv.SetSyncTrigger(syncMgr.SyncNow) + go syncMgr.Run(ctx) + return srv.Run(ctx) } diff --git a/internal/domain/status.go b/internal/domain/status.go new file mode 100644 index 0000000..6d59796 --- /dev/null +++ b/internal/domain/status.go @@ -0,0 +1,108 @@ +package domain + +import "fmt" + +// TaskStatus and its transition rules are the single source of truth for the +// §4.4 status machine. Every handler and worker validates transitions here — +// never inline. +type TaskStatus string + +const ( + StatusImported TaskStatus = "imported" + StatusAtomizing TaskStatus = "atomizing" + StatusAtomized TaskStatus = "atomized" + StatusPublished TaskStatus = "published" + StatusClaimRequested TaskStatus = "claim_requested" + StatusAssigned TaskStatus = "assigned" + StatusInProgress TaskStatus = "in_progress" + StatusInReview TaskStatus = "in_review" + StatusChangesRequested TaskStatus = "changes_requested" + StatusApproved TaskStatus = "approved" + StatusArchived TaskStatus = "archived" +) + +// AllStatuses lists every valid status (used by tests and validation). +var AllStatuses = []TaskStatus{ + StatusImported, StatusAtomizing, StatusAtomized, StatusPublished, + StatusClaimRequested, StatusAssigned, StatusInProgress, StatusInReview, + StatusChangesRequested, StatusApproved, StatusArchived, +} + +func (s TaskStatus) Valid() bool { + for _, v := range AllStatuses { + if s == v { + return true + } + } + return false +} + +// transitions encodes §4.4 exactly: +// +// imported ──subdivide──► atomizing ──service ok──► atomized (children created atomized) +// atomizing ──service failed──► imported (recovery; archived also allowed from any) +// atomized ──publish──► published +// published ──dev requests──► claim_requested ──approve──► assigned +// published ──assign AI──► assigned +// assigned ──dev starts──► in_progress ──dev submits──► in_review +// assigned ──AI callback succeeded──► in_review +// in_review ──► approved | changes_requested ──► in_progress +// claim_requested ──withdraw/decline──► published +// assigned|in_progress ──unassign/abandon──► published +// any ──► archived (terminal) +var transitions = map[TaskStatus]map[TaskStatus]bool{ + StatusImported: {StatusAtomizing: true}, + StatusAtomizing: {StatusAtomized: true, StatusImported: true}, + StatusAtomized: {StatusPublished: true, StatusAtomizing: true}, // re-run subdivision + StatusPublished: {StatusClaimRequested: true, StatusAssigned: true}, + StatusClaimRequested: { + StatusAssigned: true, + StatusPublished: true, + }, + StatusAssigned: { + StatusInProgress: true, + StatusInReview: true, // AI work-performer callback path + StatusPublished: true, + }, + StatusInProgress: { + StatusInReview: true, + StatusPublished: true, + }, + StatusInReview: { + StatusApproved: true, + StatusChangesRequested: true, + StatusAssigned: true, // AI job failed → back to assigned (§5.2) + }, + StatusChangesRequested: {StatusInProgress: true}, + StatusApproved: {}, + StatusArchived: {}, +} + +// CanTransition reports whether from → to is a legal §4.4 move. Archiving is +// allowed from any state except archived itself. +func CanTransition(from, to TaskStatus) bool { + if !from.Valid() || !to.Valid() { + return false + } + if to == StatusArchived { + return from != StatusArchived + } + return transitions[from][to] +} + +// ErrBadTransition describes a rejected move for error envelopes. +type ErrBadTransition struct { + From, To TaskStatus +} + +func (e *ErrBadTransition) Error() string { + return fmt.Sprintf("illegal status transition %s → %s", e.From, e.To) +} + +// Transition validates and returns the error form handlers translate to 409. +func Transition(from, to TaskStatus) error { + if !CanTransition(from, to) { + return &ErrBadTransition{From: from, To: to} + } + return nil +} diff --git a/internal/domain/status_test.go b/internal/domain/status_test.go new file mode 100644 index 0000000..53b2446 --- /dev/null +++ b/internal/domain/status_test.go @@ -0,0 +1,98 @@ +package domain + +import "testing" + +// TestTransitionMatrixExhaustive enumerates every (from, to) pair and pins +// the §4.4 machine exactly: the allowed set below is the complete truth. +func TestTransitionMatrixExhaustive(t *testing.T) { + allowed := map[TaskStatus]map[TaskStatus]bool{ + StatusImported: {StatusAtomizing: true, StatusArchived: true}, + StatusAtomizing: {StatusAtomized: true, StatusImported: true, StatusArchived: true}, + StatusAtomized: {StatusPublished: true, StatusAtomizing: true, StatusArchived: true}, + StatusPublished: {StatusClaimRequested: true, StatusAssigned: true, StatusArchived: true}, + StatusClaimRequested: {StatusAssigned: true, StatusPublished: true, StatusArchived: true}, + StatusAssigned: {StatusInProgress: true, StatusInReview: true, StatusPublished: true, StatusArchived: true}, + StatusInProgress: {StatusInReview: true, StatusPublished: true, StatusArchived: true}, + StatusInReview: {StatusApproved: true, StatusChangesRequested: true, StatusAssigned: true, StatusArchived: true}, + StatusChangesRequested: {StatusInProgress: true, StatusArchived: true}, + StatusApproved: {StatusArchived: true}, + StatusArchived: {}, + } + + for _, from := range AllStatuses { + for _, to := range AllStatuses { + want := allowed[from][to] + if got := CanTransition(from, to); got != want { + t.Errorf("CanTransition(%s, %s) = %v, want %v", from, to, got, want) + } + err := Transition(from, to) + if want && err != nil { + t.Errorf("Transition(%s, %s) unexpected error %v", from, to, err) + } + if !want && err == nil { + t.Errorf("Transition(%s, %s) should error", from, to) + } + } + } +} + +func TestTransitionRejectsUnknownStatuses(t *testing.T) { + if CanTransition("bogus", StatusArchived) { + t.Error("unknown from-status must be rejected") + } + if CanTransition(StatusImported, "bogus") { + t.Error("unknown to-status must be rejected") + } + if CanTransition(StatusImported, StatusImported) { + t.Error("self-transition must be rejected") + } +} + +func TestStatusValid(t *testing.T) { + for _, s := range AllStatuses { + if !s.Valid() { + t.Errorf("%s should be valid", s) + } + } + if TaskStatus("nope").Valid() { + t.Error("nope should be invalid") + } +} + +func TestComputeBounty(t *testing.T) { + tests := []struct { + coeff, budget, want float64 + }{ + {0.25, 1000, 250}, + {0.2, 1000, 200}, + {1.0 / 3.0, 1000, 333.33}, + {0.333, 100, 33.3}, + {0, 1000, 0}, + {1, 0, 0}, + {0.01, 99.99, 1.0}, // 0.9999 rounds to 1.00 + {0.005, 100, 0.5}, + {2.0, 500, 1000}, // extension coefficients may exceed 1 (§5.1) + } + for _, tt := range tests { + if got := ComputeBounty(tt.coeff, tt.budget); got != tt.want { + t.Errorf("ComputeBounty(%v, %v) = %v, want %v", tt.coeff, tt.budget, got, tt.want) + } + } +} + +func TestTaskHelpers(t *testing.T) { + task := &Task{ + Assignee: &Assignee{Kind: "human", UserID: "dev1"}, + ClaimRequests: []ClaimRequest{{DeveloperID: "dev2"}}, + } + if !task.IsAssignedTo("dev1") || task.IsAssignedTo("dev2") { + t.Error("IsAssignedTo wrong") + } + if !task.HasClaimFrom("dev2") || task.HasClaimFrom("dev1") { + t.Error("HasClaimFrom wrong") + } + aiTask := &Task{Assignee: &Assignee{Kind: "ai", JobID: "wp_1"}} + if aiTask.IsAssignedTo("dev1") { + t.Error("ai assignee must not match human checks") + } +} diff --git a/internal/domain/task.go b/internal/domain/task.go new file mode 100644 index 0000000..bf0e496 --- /dev/null +++ b/internal/domain/task.go @@ -0,0 +1,114 @@ +package domain + +import ( + "math" + "time" +) + +type TaskOrigin string + +const ( + OriginImported TaskOrigin = "imported" + OriginSubdivided TaskOrigin = "subdivided" + OriginExtended TaskOrigin = "extended" +) + +type External struct { + System string `bson:"system" json:"system"` + Key string `bson:"key" json:"key"` + URL string `bson:"url" json:"url"` + Type string `bson:"type" json:"type"` // epic | story | task (icon only) + Raw map[string]any `bson:"raw" json:"-"` + Orphaned bool `bson:"orphaned" json:"orphaned"` + ContentHash string `bson:"contentHash" json:"-"` // change detection for upstream edits +} + +type TaskAttachment struct { + Name string `bson:"name" json:"name"` + URL string `bson:"url" json:"url"` + MimeType string `bson:"mimeType" json:"mimeType"` + FileID string `bson:"fileId,omitempty" json:"fileId,omitempty"` +} + +type Assignee struct { + Kind string `bson:"kind" json:"kind"` // human | ai + UserID string `bson:"userId,omitempty" json:"userId,omitempty"` + JobID string `bson:"jobId,omitempty" json:"jobId,omitempty"` +} + +type ClaimRequest struct { + DeveloperID string `bson:"developerId" json:"developerId"` + Note string `bson:"note" json:"note"` + At time.Time `bson:"at" json:"at"` +} + +type TimelineEntry struct { + At time.Time `bson:"at" json:"at"` + ActorID string `bson:"actorId" json:"actorId"` // user id or "system" + Event string `bson:"event" json:"event"` + Data map[string]any `bson:"data,omitempty" json:"data,omitempty"` +} + +type Comment struct { + ID string `bson:"_id" json:"id"` + AuthorID string `bson:"authorId" json:"authorId"` + Body string `bson:"body" json:"body"` // sanitized HTML + At time.Time `bson:"at" json:"at"` +} + +type TimeLogEntry struct { + DeveloperID string `bson:"developerId" json:"developerId"` + Minutes int `bson:"minutes" json:"minutes"` + Note string `bson:"note" json:"note"` + At time.Time `bson:"at" json:"at"` +} + +// Task is the central work item (§4.4). +type Task struct { + ID string `bson:"_id" json:"id"` + CustomerID string `bson:"customerId" json:"customerId"` + ConsultantID string `bson:"consultantId" json:"consultantId"` + Origin TaskOrigin `bson:"origin" json:"origin"` + ParentID string `bson:"parentId,omitempty" json:"parentId,omitempty"` + RootID string `bson:"rootId" json:"rootId"` + External *External `bson:"external,omitempty" json:"external,omitempty"` + Title string `bson:"title" json:"title"` + Description string `bson:"description" json:"description"` + AcceptanceCriteria []string `bson:"acceptanceCriteria" json:"acceptanceCriteria"` + Attachments []TaskAttachment `bson:"attachments" json:"attachments"` + Links []string `bson:"links" json:"links"` + EffortCoefficient float64 `bson:"effortCoefficient" json:"effortCoefficient"` + Budget float64 `bson:"budget" json:"budget"` + Bounty float64 `bson:"bounty" json:"bounty"` + Status TaskStatus `bson:"status" json:"status"` + Assignee *Assignee `bson:"assignee,omitempty" json:"assignee,omitempty"` + ClaimRequests []ClaimRequest `bson:"claimRequests" json:"claimRequests"` + AtomizationNote string `bson:"atomizationNote,omitempty" json:"atomizationNote,omitempty"` + Timeline []TimelineEntry `bson:"timeline" json:"timeline"` + Comments []Comment `bson:"comments" json:"comments"` + TimeLog []TimeLogEntry `bson:"timeLog" json:"timeLog"` + PublishedAt time.Time `bson:"publishedAt,omitempty" json:"publishedAt,omitempty"` // stale-task aging (§11.19) + CreatedAt time.Time `bson:"createdAt" json:"createdAt"` + UpdatedAt time.Time `bson:"updatedAt" json:"updatedAt"` + Version int64 `bson:"version" json:"version"` +} + +// ComputeBounty implements §6.1: round(coefficient × budget, 2). +func ComputeBounty(coefficient, budget float64) float64 { + return math.Round(coefficient*budget*100) / 100 +} + +// IsAssignedTo reports whether the human developer holds the task. +func (t *Task) IsAssignedTo(userID string) bool { + return t.Assignee != nil && t.Assignee.Kind == "human" && t.Assignee.UserID == userID +} + +// HasClaimFrom reports an open claim request by the developer. +func (t *Task) HasClaimFrom(developerID string) bool { + for _, cr := range t.ClaimRequests { + if cr.DeveloperID == developerID { + return true + } + } + return false +} diff --git a/internal/http/providers.go b/internal/http/providers.go index 0e68dfc..4c3b5a7 100644 --- a/internal/http/providers.go +++ b/internal/http/providers.go @@ -22,6 +22,19 @@ func (s *Server) SetPerformerInfo(b BreakerInfo) { s.performer = b } // SetSyncTrigger wires the sync manager's "sync now" entry point. func (s *Server) SetSyncTrigger(f func(customerID string)) { s.syncTrigger = f } + +// Publish forwards a live event to connected clients. Until the WebSocket +// hub lands this is a metrics-only no-op, so subsystems can emit events +// unconditionally. +func (s *Server) Publish(channel, event string, payload any) { + s.metrics.Inc("events_published_total", 1) + if s.publishFn != nil { + s.publishFn(channel, event, payload) + } +} + +// SetPublishFn wires the actual hub broadcast. +func (s *Server) SetPublishFn(f func(channel, event string, payload any)) { s.publishFn = f } func (s *Server) SetSyncStatusProvider(p StatusProvider) { s.syncProvider = p } diff --git a/internal/http/server.go b/internal/http/server.go index 8c29e92..0cef178 100644 --- a/internal/http/server.go +++ b/internal/http/server.go @@ -38,6 +38,7 @@ type Server struct { syncProvider StatusProvider jobsProvider StatusProvider syncTrigger func(customerID string) + publishFn func(channel, event string, payload any) } func New(cfg *config.Config, log *slog.Logger, reg *metrics.Registry, st *store.Store, fs *files.Store) *Server { diff --git a/internal/store/notifications.go b/internal/store/notifications.go new file mode 100644 index 0000000..e74136f --- /dev/null +++ b/internal/store/notifications.go @@ -0,0 +1,79 @@ +package store + +import ( + "context" + "fmt" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo/options" + + "bountyboard/internal/ulid" +) + +// Notification is an in-app notification (§4.8). +type Notification struct { + ID string `bson:"_id" json:"id"` + UserID string `bson:"userId" json:"userId"` + Kind string `bson:"kind" json:"kind"` + Title string `bson:"title" json:"title"` + Body string `bson:"body" json:"body"` + Link string `bson:"link" json:"link"` + ReadAt *time.Time `bson:"readAt" json:"readAt"` + CreatedAt time.Time `bson:"createdAt" json:"createdAt"` +} + +func (s *Store) InsertNotification(ctx context.Context, n *Notification) error { + n.ID = ulid.New() + n.CreatedAt = time.Now().UTC() + if _, err := s.DB.Collection("notifications").InsertOne(ctx, n); err != nil { + return fmt.Errorf("insert notification: %w", err) + } + return nil +} + +func (s *Store) ListNotifications(ctx context.Context, userID string, unreadOnly bool, cursor string, limit int) ([]Notification, error) { + if limit <= 0 || limit > 100 { + limit = 30 + } + q := bson.M{"userId": userID} + if unreadOnly { + q["readAt"] = nil + } + if cursor != "" { + q["_id"] = bson.M{"$lt": cursor} + } + cur, err := s.DB.Collection("notifications").Find(ctx, q, + options.Find().SetSort(bson.D{{Key: "_id", Value: -1}}).SetLimit(int64(limit))) + if err != nil { + return nil, fmt.Errorf("list notifications: %w", err) + } + out := []Notification{} + if err := cur.All(ctx, &out); err != nil { + return nil, fmt.Errorf("decode notifications: %w", err) + } + return out, nil +} + +func (s *Store) CountUnreadNotifications(ctx context.Context, userID string) (int64, error) { + n, err := s.DB.Collection("notifications").CountDocuments(ctx, + bson.M{"userId": userID, "readAt": nil}) + if err != nil { + return 0, fmt.Errorf("count unread: %w", err) + } + return n, nil +} + +// MarkNotificationsRead marks the given ids (or all when ids is empty) read. +func (s *Store) MarkNotificationsRead(ctx context.Context, userID string, ids []string) (int64, error) { + q := bson.M{"userId": userID, "readAt": nil} + if len(ids) > 0 { + q["_id"] = bson.M{"$in": ids} + } + res, err := s.DB.Collection("notifications").UpdateMany(ctx, q, + bson.M{"$set": bson.M{"readAt": time.Now().UTC()}}) + if err != nil { + return 0, fmt.Errorf("mark read: %w", err) + } + return res.ModifiedCount, nil +} diff --git a/internal/store/tasks.go b/internal/store/tasks.go new file mode 100644 index 0000000..d46913e --- /dev/null +++ b/internal/store/tasks.go @@ -0,0 +1,330 @@ +package store + +import ( + "context" + "errors" + "fmt" + "time" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" + + "bountyboard/internal/domain" + "bountyboard/internal/ulid" +) + +func (s *Store) tasks() *mongo.Collection { return s.DB.Collection("tasks") } + +func (s *Store) InsertTask(ctx context.Context, t *domain.Task) error { + now := time.Now().UTC() + if t.ID == "" { + t.ID = ulid.New() + } + if t.RootID == "" { + t.RootID = t.ID + } + t.CreatedAt, t.UpdatedAt, t.Version = now, now, 1 + if t.AcceptanceCriteria == nil { + t.AcceptanceCriteria = []string{} + } + if t.Attachments == nil { + t.Attachments = []domain.TaskAttachment{} + } + if t.Links == nil { + t.Links = []string{} + } + if t.ClaimRequests == nil { + t.ClaimRequests = []domain.ClaimRequest{} + } + if t.Timeline == nil { + t.Timeline = []domain.TimelineEntry{} + } + if t.Comments == nil { + t.Comments = []domain.Comment{} + } + if t.TimeLog == nil { + t.TimeLog = []domain.TimeLogEntry{} + } + if _, err := s.tasks().InsertOne(ctx, t); err != nil { + if mongo.IsDuplicateKeyError(err) { + return fmt.Errorf("task %s: %w", t.ID, ErrDuplicate) + } + return fmt.Errorf("insert task: %w", err) + } + return nil +} + +func (s *Store) TaskByID(ctx context.Context, id string) (*domain.Task, error) { + var t domain.Task + err := s.tasks().FindOne(ctx, bson.M{"_id": id}).Decode(&t) + if errors.Is(err, mongo.ErrNoDocuments) { + return nil, ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("find task: %w", err) + } + return &t, nil +} + +// TaskFilter selects tasks for boards and workers. +type TaskFilter struct { + CustomerID string + CustomerIDs []string + ConsultantID string + ConsultantIDs []string + Statuses []domain.TaskStatus + AssigneeID string + RootID string + ParentID string + Search string // Mongo text index + MinBounty float64 + Sort string // newest | bounty | updated + Cursor string + Limit int +} + +func (f TaskFilter) build() bson.M { + q := bson.M{} + if f.CustomerID != "" { + q["customerId"] = f.CustomerID + } + if len(f.CustomerIDs) > 0 { + q["customerId"] = bson.M{"$in": f.CustomerIDs} + } + if f.ConsultantID != "" { + q["consultantId"] = f.ConsultantID + } + if len(f.ConsultantIDs) > 0 { + q["consultantId"] = bson.M{"$in": f.ConsultantIDs} + } + if len(f.Statuses) > 0 { + q["status"] = bson.M{"$in": f.Statuses} + } + if f.AssigneeID != "" { + q["assignee.userId"] = f.AssigneeID + } + if f.RootID != "" { + q["rootId"] = f.RootID + } + if f.ParentID != "" { + q["parentId"] = f.ParentID + } + if f.Search != "" { + q["$text"] = bson.M{"$search": f.Search} + } + if f.MinBounty > 0 { + q["bounty"] = bson.M{"$gte": f.MinBounty} + } + return q +} + +func (s *Store) ListTasks(ctx context.Context, f TaskFilter) ([]domain.Task, error) { + q := f.build() + limit := f.Limit + if limit <= 0 || limit > 500 { + limit = 100 + } + sort := bson.D{{Key: "_id", Value: -1}} // newest first (ULID) + switch f.Sort { + case "bounty": + sort = bson.D{{Key: "bounty", Value: -1}, {Key: "_id", Value: -1}} + case "updated": + sort = bson.D{{Key: "updatedAt", Value: -1}} + } + if f.Cursor != "" && f.Sort != "bounty" && f.Sort != "updated" { + q["_id"] = bson.M{"$lt": f.Cursor} + } + cur, err := s.tasks().Find(ctx, q, options.Find().SetSort(sort).SetLimit(int64(limit))) + if err != nil { + return nil, fmt.Errorf("list tasks: %w", err) + } + out := []domain.Task{} + if err := cur.All(ctx, &out); err != nil { + return nil, fmt.Errorf("decode tasks: %w", err) + } + return out, nil +} + +// UpsertResult describes what an imported-ticket upsert did. +type UpsertResult struct { + Created bool + UpstreamChanged bool // changed after atomization began (timeline-only) + Task *domain.Task +} + +// UpsertImportedTask implements the §5.3 idempotent upsert keyed on +// (external.system, external.key, customerId). While the task is still +// `imported`, upstream refreshes title/description/attachments/raw; after +// atomization begins, a change only appends an "upstream ticket changed" +// timeline entry (caller notifies the consultant). +func (s *Store) UpsertImportedTask(ctx context.Context, incoming *domain.Task) (*UpsertResult, error) { + if incoming.External == nil { + return nil, errors.New("imported task requires external ref") + } + key := bson.M{ + "external.system": incoming.External.System, + "external.key": incoming.External.Key, + "customerId": incoming.CustomerID, + } + var existing domain.Task + err := s.tasks().FindOne(ctx, key).Decode(&existing) + switch { + case errors.Is(err, mongo.ErrNoDocuments): + incoming.Origin = domain.OriginImported + incoming.Status = domain.StatusImported + incoming.Timeline = []domain.TimelineEntry{{ + At: time.Now().UTC(), ActorID: "system", Event: "imported", + Data: map[string]any{"key": incoming.External.Key}, + }} + if err := s.InsertTask(ctx, incoming); err != nil { + // A concurrent worker may have inserted the same key (unique + // index): treat as found-and-unchanged. + if errors.Is(err, ErrDuplicate) { + return &UpsertResult{Created: false, Task: incoming}, nil + } + return nil, err + } + return &UpsertResult{Created: true, Task: incoming}, nil + case err != nil: + return nil, fmt.Errorf("find imported task: %w", err) + } + + if existing.External.ContentHash == incoming.External.ContentHash { + return &UpsertResult{Task: &existing}, nil // nothing new upstream + } + + now := time.Now().UTC() + if existing.Status == domain.StatusImported { + _, err := s.tasks().UpdateOne(ctx, bson.M{"_id": existing.ID}, bson.M{ + "$set": bson.M{ + "title": incoming.Title, + "description": incoming.Description, + "attachments": incoming.Attachments, + "external.raw": incoming.External.Raw, + "external.url": incoming.External.URL, + "external.type": incoming.External.Type, + "external.contentHash": incoming.External.ContentHash, + "updatedAt": now, + }, + "$inc": bson.M{"version": 1}, + }) + if err != nil { + return nil, fmt.Errorf("refresh imported task: %w", err) + } + return &UpsertResult{Task: &existing}, nil + } + + // Atomization already began: record only. + _, err = s.tasks().UpdateOne(ctx, bson.M{"_id": existing.ID}, bson.M{ + "$set": bson.M{"external.contentHash": incoming.External.ContentHash, "updatedAt": now}, + "$push": bson.M{"timeline": domain.TimelineEntry{ + At: now, ActorID: "system", Event: "upstream_changed", + Data: map[string]any{"key": incoming.External.Key}, + }}, + "$inc": bson.M{"version": 1}, + }) + if err != nil { + return nil, fmt.Errorf("record upstream change: %w", err) + } + return &UpsertResult{UpstreamChanged: true, Task: &existing}, nil +} + +// MarkTaskOrphaned flags a task whose upstream ticket disappeared (§5.3). +func (s *Store) MarkTaskOrphaned(ctx context.Context, id string) error { + now := time.Now().UTC() + res, err := s.tasks().UpdateOne(ctx, + bson.M{"_id": id, "external.orphaned": bson.M{"$ne": true}}, + bson.M{ + "$set": bson.M{"external.orphaned": true, "updatedAt": now}, + "$push": bson.M{"timeline": domain.TimelineEntry{ + At: now, ActorID: "system", Event: "orphaned", + Data: map[string]any{"reason": "no longer returned by upstream query"}, + }}, + "$inc": bson.M{"version": 1}, + }) + if err != nil { + return fmt.Errorf("mark orphaned: %w", err) + } + if res.MatchedCount == 0 { + return ErrNotFound // already orphaned or missing + } + return nil +} + +// AppendTimeline pushes a timeline entry without status change. +func (s *Store) AppendTimeline(ctx context.Context, taskID string, e domain.TimelineEntry) error { + if e.At.IsZero() { + e.At = time.Now().UTC() + } + _, err := s.tasks().UpdateOne(ctx, bson.M{"_id": taskID}, bson.M{ + "$push": bson.M{"timeline": e}, + "$set": bson.M{"updatedAt": time.Now().UTC()}, + "$inc": bson.M{"version": 1}, + }) + if err != nil { + return fmt.Errorf("append timeline: %w", err) + } + return nil +} + +// TransitionTask performs a version-checked §4.4 transition with a timeline +// entry and optional extra $set fields. +func (s *Store) TransitionTask(ctx context.Context, t *domain.Task, to domain.TaskStatus, + actorID, event string, data map[string]any, extraSet bson.M) error { + if err := domain.Transition(t.Status, to); err != nil { + return err + } + set := bson.M{"status": to} + for k, v := range extraSet { + set[k] = v + } + update := bson.M{ + "$set": set, + "$push": bson.M{"timeline": domain.TimelineEntry{ + At: time.Now().UTC(), ActorID: actorID, Event: event, Data: data, + }}, + } + return UpdateVersioned(ctx, s.tasks(), t.ID, t.Version, update) +} + +// SetTaskAttachments persists attachment metadata (e.g. cached fileIds). +func (s *Store) SetTaskAttachments(ctx context.Context, taskID string, atts []domain.TaskAttachment) error { + _, err := s.tasks().UpdateOne(ctx, bson.M{"_id": taskID}, bson.M{ + "$set": bson.M{"attachments": atts, "updatedAt": time.Now().UTC()}, + "$inc": bson.M{"version": 1}, + }) + if err != nil { + return fmt.Errorf("set attachments: %w", err) + } + return nil +} + +// ListImportedTasksForOrphanCheck returns imported-origin tasks for one +// customer/system/consultant that are still live (not archived). +func (s *Store) ListImportedTasksForOrphanCheck(ctx context.Context, customerID, system, consultantID string) ([]domain.Task, error) { + cur, err := s.tasks().Find(ctx, bson.M{ + "customerId": customerID, + "consultantId": consultantID, + "origin": domain.OriginImported, + "external.system": system, + "status": bson.M{"$ne": domain.StatusArchived}, + }) + if err != nil { + return nil, fmt.Errorf("list for orphan check: %w", err) + } + out := []domain.Task{} + if err := cur.All(ctx, &out); err != nil { + return nil, fmt.Errorf("decode orphan check: %w", err) + } + return out, nil +} + +// DeleteTasks removes tasks matching ids (re-run subdivision cleanup). +func (s *Store) DeleteTasks(ctx context.Context, ids []string) (int64, error) { + res, err := s.tasks().DeleteMany(ctx, bson.M{"_id": bson.M{"$in": ids}}) + if err != nil { + return 0, fmt.Errorf("delete tasks: %w", err) + } + return res.DeletedCount, nil +} diff --git a/internal/store/users.go b/internal/store/users.go index 8a1338b..589623a 100644 --- a/internal/store/users.go +++ b/internal/store/users.go @@ -115,6 +115,23 @@ func (s *Store) LinkOIDC(ctx context.Context, userID, issuer, subject string) er return nil } +// ListAdminIDs returns ids of all active admins (system notifications). +func (s *Store) ListAdminIDs(ctx context.Context) ([]string, error) { + cur, err := s.users().Find(ctx, bson.M{"roles.admin": true, "disabled": false}) + if err != nil { + return nil, fmt.Errorf("list admins: %w", err) + } + var users []domain.User + if err := cur.All(ctx, &users); err != nil { + return nil, fmt.Errorf("decode admins: %w", err) + } + ids := make([]string, len(users)) + for i, u := range users { + ids[i] = u.ID + } + return ids, nil +} + // TouchLastSeen updates lastSeenAt without bumping version (pure telemetry). func (s *Store) TouchLastSeen(ctx context.Context, userID string) error { _, err := s.users().UpdateOne(ctx, bson.M{"_id": userID}, diff --git a/internal/sync/manager.go b/internal/sync/manager.go new file mode 100644 index 0000000..f345604 --- /dev/null +++ b/internal/sync/manager.go @@ -0,0 +1,446 @@ +package sync + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "log/slog" + "net/http" + "sync" + "time" + + "bountyboard/internal/domain" + "bountyboard/internal/files" + "bountyboard/internal/metrics" + "bountyboard/internal/store" +) + +// overlapWindow re-fetches a little history each poll so clock skew or slow +// writes can't drop updates (§5.3). +const overlapWindow = 5 * time.Minute + +// Publisher pushes live updates (wired to the WS hub when it exists). +type Publisher func(channel, event string, payload any) + +// Manager runs one polling worker per syncable customer and reconciles the +// worker set against the customers collection. +type Manager struct { + st *store.Store + files *files.Store + log *slog.Logger + reg *metrics.Registry + creds func(*domain.Customer) (Credentials, error) + publish Publisher + + mu sync.Mutex + workers map[string]*worker +} + +type worker struct { + customerID string + interval time.Duration + trigger chan struct{} + cancel context.CancelFunc + + mu sync.Mutex + name string + lastStatus string + lastError string + lastRunAt time.Time +} + +// WorkerStatus feeds the admin service-status panel (§11.17). +type WorkerStatus struct { + CustomerID string `json:"customerId"` + CustomerName string `json:"customerName"` + PollIntervalSec int `json:"pollIntervalSec"` + LastStatus string `json:"lastStatus"` + LastError string `json:"lastError,omitempty"` + LastRunAt time.Time `json:"lastRunAt"` +} + +func NewManager(st *store.Store, fs *files.Store, log *slog.Logger, reg *metrics.Registry, + creds func(*domain.Customer) (Credentials, error), publish Publisher) *Manager { + if publish == nil { + publish = func(string, string, any) {} + } + return &Manager{ + st: st, files: fs, log: log, reg: reg, creds: creds, publish: publish, + workers: map[string]*worker{}, + } +} + +// Run reconciles workers until ctx is done, then stops them all (graceful +// shutdown flushes by waiting for the in-flight tick via context). +func (m *Manager) Run(ctx context.Context) { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + m.reconcile(ctx) + for { + select { + case <-ctx.Done(): + m.mu.Lock() + for _, w := range m.workers { + w.cancel() + } + m.mu.Unlock() + return + case <-ticker.C: + m.reconcile(ctx) + } + } +} + +func (m *Manager) reconcile(ctx context.Context) { + customers, err := m.st.ListCustomers(ctx, "", false) + if err != nil { + m.log.Error("sync reconcile: list customers", "err", err) + return + } + m.mu.Lock() + defer m.mu.Unlock() + + want := map[string]*domain.Customer{} + for i := range customers { + c := &customers[i] + if c.Ticketing.Type.Valid() && len(c.ConsultantIDs) > 0 { + want[c.ID] = c + } + } + // stop removed/archived customers + for id, w := range m.workers { + if _, ok := want[id]; !ok { + w.cancel() + delete(m.workers, id) + } + } + // start new / restart changed-interval workers + for id, c := range want { + interval := time.Duration(c.Ticketing.PollIntervalSec) * time.Second + if interval < 10*time.Second { + interval = time.Minute + } + if w, ok := m.workers[id]; ok { + if w.interval == interval { + w.mu.Lock() + w.name = c.Name + w.mu.Unlock() + continue + } + w.cancel() + delete(m.workers, id) + } + wctx, cancel := context.WithCancel(ctx) + w := &worker{ + customerID: id, interval: interval, name: c.Name, + trigger: make(chan struct{}, 1), cancel: cancel, + } + m.workers[id] = w + go m.runWorker(wctx, w) + } + m.reg.Set("sync_workers", int64(len(m.workers))) +} + +// SyncNow triggers an immediate poll for the customer. +func (m *Manager) SyncNow(customerID string) { + m.mu.Lock() + w, ok := m.workers[customerID] + m.mu.Unlock() + if ok { + select { + case w.trigger <- struct{}{}: + default: // already queued + } + } +} + +// Statuses implements the admin panel provider. +func (m *Manager) Statuses() any { + m.mu.Lock() + defer m.mu.Unlock() + out := []WorkerStatus{} + for _, w := range m.workers { + w.mu.Lock() + out = append(out, WorkerStatus{ + CustomerID: w.customerID, + CustomerName: w.name, + PollIntervalSec: int(w.interval.Seconds()), + LastStatus: w.lastStatus, + LastError: w.lastError, + LastRunAt: w.lastRunAt, + }) + w.mu.Unlock() + } + return out +} + +func (m *Manager) runWorker(ctx context.Context, w *worker) { + // first run promptly, then on the interval + timer := time.NewTimer(2 * time.Second) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + case <-w.trigger: + } + m.syncOnce(ctx, w) + timer.Reset(w.interval) + } +} + +func (m *Manager) syncOnce(ctx context.Context, w *worker) { + defer func() { + if rec := recover(); rec != nil { + m.log.Error("sync worker panic", "customerId", w.customerID, "panic", fmt.Sprint(rec)) + } + }() + ctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + + c, err := m.st.CustomerByID(ctx, w.customerID) + if err != nil { + m.recordRun(ctx, w, nil, err) + return + } + err = m.SyncCustomer(ctx, c) + m.recordRun(ctx, w, c, err) +} + +func (m *Manager) recordRun(ctx context.Context, w *worker, c *domain.Customer, err error) { + now := time.Now().UTC() + status, msg := "ok", "" + if err != nil { + status, msg = "error", err.Error() + m.log.Error("sync run failed", "customerId", w.customerID, "err", err) + m.reg.Inc("sync_runs_failed_total", 1) + // surface as admin notification (§12) — best effort, throttled by + // only notifying on status flips + if c != nil && c.Ticketing.LastSyncStatus != "error" { + m.notifyAdmins(ctx, "sync_error", "Sync failed: "+c.Name, msg, "/admin") + } + } else { + m.reg.Inc("sync_runs_total", 1) + } + w.mu.Lock() + w.lastStatus, w.lastError, w.lastRunAt = status, msg, now + w.mu.Unlock() + if setErr := m.st.SetSyncStatus(ctx, w.customerID, now, status, msg); setErr != nil { + m.log.Warn("set sync status", "err", setErr) + } +} + +// SyncCustomer performs one full sync pass for a customer; exported for +// tests and the sync-now path. +func (m *Manager) SyncCustomer(ctx context.Context, c *domain.Customer) error { + creds, err := m.creds(c) + if err != nil { + return fmt.Errorf("decrypt credentials: %w", err) + } + conn, err := NewConnector(c.Ticketing.Type, c.Ticketing.BaseURL, c.Ticketing.ProjectKey, creds) + if err != nil { + return err + } + + since := c.Ticketing.LastSyncAt.Add(-overlapWindow) + if c.Ticketing.LastSyncAt.IsZero() { + since = time.Unix(0, 0) + } + + var firstErr error + for _, consultantID := range c.ConsultantIDs { + consultant, err := m.st.UserByID(ctx, consultantID) + if err != nil { + continue // consultant deleted; reconcile will catch up + } + identity := ticketingIdentity(consultant, c.Ticketing.Type) + if identity == "" { + continue // §5.3: consultant has no linked account in this system + } + + tickets, err := conn.FetchUpdated(ctx, identity, since) + if err != nil { + if firstErr == nil { + firstErr = fmt.Errorf("fetch for %s: %w", consultant.Email, err) + } + continue + } + for _, ticket := range tickets { + if err := m.upsertTicket(ctx, c, consultant, conn, ticket); err != nil { + m.log.Error("upsert ticket", "key", ticket.Key, "err", err) + if firstErr == nil { + firstErr = err + } + } + } + + if err := m.flagOrphans(ctx, c, consultantID, conn, identity); err != nil { + m.log.Warn("orphan detection", "customerId", c.ID, "err", err) + } + } + return firstErr +} + +func ticketingIdentity(u *domain.User, t domain.TicketingType) string { + if t == domain.TicketingDemo { + return u.Email // demo accepts any identity + } + ids, _ := u.Extra["ticketingIdentities"].(map[string]any) + if ids == nil { + return "" + } + v, _ := ids[string(t)].(string) + return v +} + +func contentHash(t Ticket) string { + h := sha256.New() + fmt.Fprintf(h, "%s|%s|%s|%d", t.Title, t.Description, t.Type, len(t.Attachments)) + for _, a := range t.Attachments { + fmt.Fprintf(h, "|%s", a.Name) + } + return hex.EncodeToString(h.Sum(nil)) +} + +func (m *Manager) upsertTicket(ctx context.Context, c *domain.Customer, + consultant *domain.User, conn Connector, ticket Ticket) error { + + task := &domain.Task{ + CustomerID: c.ID, + ConsultantID: consultant.ID, + Title: ticket.Title, + Description: ticket.Description, + Budget: c.DefaultBudget, // §11.20 pre-fill + EffortCoefficient: 1, + External: &domain.External{ + System: string(c.Ticketing.Type), + Key: ticket.Key, + URL: ticket.URL, + Type: ticket.Type, + Raw: ticket.Raw, + ContentHash: contentHash(ticket), + }, + } + task.Bounty = domain.ComputeBounty(task.EffortCoefficient, task.Budget) + for _, a := range ticket.Attachments { + task.Attachments = append(task.Attachments, domain.TaskAttachment{ + Name: a.Name, URL: a.URL, MimeType: a.MimeType, + }) + } + + res, err := m.st.UpsertImportedTask(ctx, task) + if err != nil { + return err + } + switch { + case res.Created: + m.reg.Inc("sync_tickets_imported_total", 1) + m.cacheAttachments(ctx, conn, res.Task) + m.notify(ctx, consultant.ID, "ticket_imported", + "New ticket imported: "+ticket.Key, + ticket.Title, "/tasks/"+res.Task.ID) + m.publish("board", "task.imported", map[string]any{"taskId": res.Task.ID, "customerId": c.ID}) + case res.UpstreamChanged: + m.notify(ctx, consultant.ID, "upstream_changed", + "Upstream ticket changed: "+ticket.Key, + "The source ticket changed after atomization began.", "/tasks/"+res.Task.ID) + } + return nil +} + +// cacheAttachments downloads ticket attachments into GridFS so the atomizer +// can fetch them from the app (§5.3); failures are logged, not fatal. +func (m *Manager) cacheAttachments(ctx context.Context, conn Connector, t *domain.Task) { + changed := false + for i, a := range t.Attachments { + if a.FileID != "" || a.URL == "" { + continue + } + fileID, err := m.downloadAttachment(ctx, conn, t.ID, a) + if err != nil { + m.log.Warn("cache attachment", "task", t.ID, "name", a.Name, "err", err) + continue + } + t.Attachments[i].FileID = fileID + changed = true + } + if changed { + if err := m.st.SetTaskAttachments(ctx, t.ID, t.Attachments); err != nil { + m.log.Warn("persist attachment cache", "task", t.ID, "err", err) + } + } +} + +func (m *Manager) downloadAttachment(ctx context.Context, conn Connector, taskID string, a domain.TaskAttachment) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.URL, nil) + if err != nil { + return "", err + } + conn.Authorize(req) + resp, err := httpClient.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("download %s: status %d", a.URL, resp.StatusCode) + } + meta, err := m.files.Save(ctx, files.ScopeTask, "system:"+taskID, a.Name, a.MimeType, resp.Body) + if err != nil { + return "", err + } + return meta.ID, nil +} + +func (m *Manager) flagOrphans(ctx context.Context, c *domain.Customer, + consultantID string, conn Connector, identity string) error { + + keys, err := conn.ListAssignedKeys(ctx, identity) + if err != nil { + return err + } + current := make(map[string]bool, len(keys)) + for _, k := range keys { + current[k] = true + } + tasks, err := m.st.ListImportedTasksForOrphanCheck(ctx, c.ID, string(c.Ticketing.Type), consultantID) + if err != nil { + return err + } + for _, t := range tasks { + if t.External == nil || current[t.External.Key] || t.External.Orphaned { + continue + } + if err := m.st.MarkTaskOrphaned(ctx, t.ID); err != nil { + continue // raced with another worker — fine + } + m.reg.Inc("sync_tickets_orphaned_total", 1) + m.notify(ctx, consultantID, "ticket_orphaned", + "Ticket disappeared upstream: "+t.External.Key, + t.Title+" — decide whether to archive it.", "/tasks/"+t.ID) + m.publish("board", "task.orphaned", map[string]any{"taskId": t.ID, "customerId": c.ID}) + } + return nil +} + +func (m *Manager) notify(ctx context.Context, userID, kind, title, body, link string) { + n := &store.Notification{UserID: userID, Kind: kind, Title: title, Body: body, Link: link} + if err := m.st.InsertNotification(ctx, n); err != nil { + m.log.Warn("insert notification", "err", err) + return + } + m.publish("notifications", "notification", n) +} + +func (m *Manager) notifyAdmins(ctx context.Context, kind, title, body, link string) { + admins, err := m.st.ListAdminIDs(ctx) + if err != nil { + m.log.Warn("list admins", "err", err) + return + } + for _, id := range admins { + m.notify(ctx, id, kind, title, body, link) + } +} diff --git a/internal/sync/manager_integration_test.go b/internal/sync/manager_integration_test.go new file mode 100644 index 0000000..62e08d6 --- /dev/null +++ b/internal/sync/manager_integration_test.go @@ -0,0 +1,228 @@ +//go:build integration + +package sync + +import ( + "context" + "io" + "log/slog" + "testing" + + "go.mongodb.org/mongo-driver/v2/bson" + + "bountyboard/internal/domain" + "bountyboard/internal/files" + "bountyboard/internal/metrics" + "bountyboard/internal/store" + "bountyboard/internal/testutil" + "bountyboard/internal/ulid" +) + +func newSyncStack(t *testing.T) (*Manager, *store.Store, *domain.Customer, *domain.User) { + t.Helper() + db := testutil.MongoDB(t) + if err := store.EnsureIndexes(t.Context(), db); err != nil { + t.Fatal(err) + } + st := &store.Store{Client: db.Client(), DB: db} + + consultant := &domain.User{ + ID: ulid.New(), Email: "cons@example.com", Name: "Cons", + Roles: domain.Roles{Consultant: true}, + } + if err := st.CreateUser(t.Context(), consultant); err != nil { + t.Fatal(err) + } + customer := &domain.Customer{ + Name: "Demo Co", + Ticketing: domain.Ticketing{ + Type: domain.TicketingDemo, ProjectKey: "DEMO", PollIntervalSec: 60, + }, + ConsultantIDs: []string{consultant.ID}, + DefaultBudget: 1500, + } + if err := st.CreateCustomer(t.Context(), customer); err != nil { + t.Fatal(err) + } + + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + m := NewManager(st, files.NewStore(db, 25), log, metrics.NewRegistry(), + func(*domain.Customer) (Credentials, error) { return Credentials{}, nil }, nil) + return m, st, customer, consultant +} + +func countTasks(t *testing.T, st *store.Store, filter bson.M) int64 { + t.Helper() + n, err := st.DB.Collection("tasks").CountDocuments(t.Context(), filter) + if err != nil { + t.Fatal(err) + } + return n +} + +func TestDemoSyncImportsAndIsIdempotent(t *testing.T) { + m, st, customer, consultant := newSyncStack(t) + ctx := context.Background() + + if err := m.SyncCustomer(ctx, customer); err != nil { + t.Fatalf("first sync: %v", err) + } + if n := countTasks(t, st, bson.M{"customerId": customer.ID}); n != 5 { + t.Fatalf("after first sync: %d tasks, want 5", n) + } + + // idempotent: same tickets re-fetched (overlap window) don't duplicate + for i := 0; i < 3; i++ { + if err := m.SyncCustomer(ctx, customer); err != nil { + t.Fatalf("re-sync %d: %v", i, err) + } + } + if n := countTasks(t, st, bson.M{"customerId": customer.ID}); n != 5 { + t.Fatalf("after re-syncs: %d tasks, want 5", n) + } + + // imported tasks carry the right shape + tasks, err := st.ListTasks(ctx, store.TaskFilter{CustomerID: customer.ID}) + if err != nil { + t.Fatal(err) + } + for _, task := range tasks { + if task.Status != domain.StatusImported || task.Origin != domain.OriginImported { + t.Errorf("task %s: status=%s origin=%s", task.ID, task.Status, task.Origin) + } + if task.ConsultantID != consultant.ID || task.RootID != task.ID { + t.Errorf("task %s: consultant=%s root=%s", task.ID, task.ConsultantID, task.RootID) + } + if task.Budget != 1500 { + t.Errorf("task %s: budget %v, want customer default 1500", task.ID, task.Budget) + } + if task.External == nil || task.External.System != "demo" { + t.Errorf("task %s: external %+v", task.ID, task.External) + } + } + + // consultant got import notifications (5, once each) + notifs, err := st.ListNotifications(ctx, consultant.ID, false, "", 100) + if err != nil { + t.Fatal(err) + } + imported := 0 + for _, n := range notifs { + if n.Kind == "ticket_imported" { + imported++ + } + } + if imported != 5 { + t.Fatalf("import notifications = %d, want 5", imported) + } +} + +func TestSyncPreservesWorkAfterAtomization(t *testing.T) { + m, st, customer, _ := newSyncStack(t) + ctx := context.Background() + if err := m.SyncCustomer(ctx, customer); err != nil { + t.Fatal(err) + } + tasks, _ := st.ListTasks(ctx, store.TaskFilter{CustomerID: customer.ID}) + target := tasks[0] + + // simulate atomization started + consultant edits + if _, err := st.DB.Collection("tasks").UpdateOne(ctx, bson.M{"_id": target.ID}, + bson.M{"$set": bson.M{ + "status": domain.StatusAtomized, "title": "Consultant Edited Title", + "external.contentHash": "different-upstream-hash", + }}); err != nil { + t.Fatal(err) + } + + if err := m.SyncCustomer(ctx, customer); err != nil { + t.Fatal(err) + } + fresh, err := st.TaskByID(ctx, target.ID) + if err != nil { + t.Fatal(err) + } + if fresh.Title != "Consultant Edited Title" { + t.Fatalf("title clobbered after atomization: %q", fresh.Title) + } + var sawUpstreamChange bool + for _, e := range fresh.Timeline { + if e.Event == "upstream_changed" { + sawUpstreamChange = true + } + } + if !sawUpstreamChange { + t.Fatal("timeline missing upstream_changed entry") + } +} + +func TestOrphanFlagging(t *testing.T) { + m, st, customer, consultant := newSyncStack(t) + ctx := context.Background() + if err := m.SyncCustomer(ctx, customer); err != nil { + t.Fatal(err) + } + + // flip the demo project key to the orphan variant: DEMO-5 disappears + if _, err := st.DB.Collection("customers").UpdateOne(ctx, bson.M{"_id": customer.ID}, + bson.M{"$set": bson.M{"ticketing.projectKey": "DEMO-ORPHAN"}}); err != nil { + t.Fatal(err) + } + fresh, err := st.CustomerByID(ctx, customer.ID) + if err != nil { + t.Fatal(err) + } + if err := m.SyncCustomer(ctx, fresh); err != nil { + t.Fatal(err) + } + + orphaned := countTasks(t, st, bson.M{"customerId": customer.ID, "external.orphaned": true}) + if orphaned != 1 { + t.Fatalf("orphaned = %d, want 1", orphaned) + } + // the task is NOT deleted (§5.3) + if n := countTasks(t, st, bson.M{"customerId": customer.ID}); n != 5 { + t.Fatalf("tasks after orphan = %d, want 5 (never deleted)", n) + } + + // consultant notified once, and re-sync doesn't duplicate the flagging + if err := m.SyncCustomer(ctx, fresh); err != nil { + t.Fatal(err) + } + notifs, _ := st.ListNotifications(ctx, consultant.ID, false, "", 100) + orphanNotifs := 0 + for _, n := range notifs { + if n.Kind == "ticket_orphaned" { + orphanNotifs++ + } + } + if orphanNotifs != 1 { + t.Fatalf("orphan notifications = %d, want exactly 1", orphanNotifs) + } +} + +func TestSyncSkipsConsultantWithoutIdentity(t *testing.T) { + _, st, customer, _ := newSyncStack(t) + ctx := context.Background() + + // switch to jira: consultant has no jira identity in extra + if _, err := st.DB.Collection("customers").UpdateOne(ctx, bson.M{"_id": customer.ID}, + bson.M{"$set": bson.M{"ticketing.type": "jira", "ticketing.baseUrl": "https://nope.invalid"}}); err != nil { + t.Fatal(err) + } + fresh, _ := st.CustomerByID(ctx, customer.ID) + + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + m := NewManager(st, files.NewStore(st.DB, 25), log, metrics.NewRegistry(), + func(*domain.Customer) (Credentials, error) { + return Credentials{"email": "c@x.y", "apiToken": "token"}, nil + }, nil) + + // must not error: the consultant is skipped, nothing is ever fetched + if err := m.SyncCustomer(ctx, fresh); err != nil { + t.Fatalf("sync with missing identity should be a no-op, got %v", err) + } + if n := countTasks(t, st, bson.M{"customerId": customer.ID}); n != 0 { + t.Fatalf("tasks = %d, want 0", n) + } +}