From 9dc679e786dfd9ea2f85c004ac2c840a87b62073 Mon Sep 17 00:00:00 2001 From: xarmian Date: Sun, 18 Oct 2026 12:12:51 -0400 Subject: [PATCH] feat(server): per-user catch-up digest and read cursor (TASK-3615, PR A) (#2219) * wip(store): workspace_read_cursors table and cursor store (TASK-3615, PR A part 1) The per-user, per-workspace read cursor behind the catch-up card. seen_at moves only on Mark caught up (Dave, 10-17) and never backwards; reading never writes (lead ruling, day 12). Untested checkpoint commit; the digest endpoint, PUT /me/seen and tests follow. Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * wip(server): catch-up digest store query, GET /catch-up and PUT /me/seen (TASK-3615, PR A part 2) Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * test(server): catch-up digest sections, window, cursor, visibility, needs_you, agents (TASK-3615, PR A) Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * fix(store): read activity metadata as text on Postgres in the catch-up queries (TASK-3615) Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * test(server): catch-up fixtures for reopen, abandoned, own and unrelated comments, old cursor, done-but-blocked, scoped needs_you (TASK-3615) Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * fix(server): needs_you skips reference items; classify the catch-up and blocker lifecycle sites (TASK-3615, PLAN-3535) Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * fix(store): catch-up window's lower bound is inclusive, so a same-second event is re-shown, never hidden; document the endpoints (TASK-3615) Review round 1 (hunter, Low): an event stamped in the cursor's second could have happened after the press and was never shown. Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ * test(server): classify GET /catch-up as compress in the route census (TASK-3615) Its response carries refs, titles, display names and timestamps: no token, code, signed URL or secret. Claude-Session: https://claude.ai/code/session_01HsAbPwmN6oyEuQAp6z6NhQ --- docs/api.md | 2 + .../plan3535_lifecycle_sites_guard_test.go | 2 + internal/server/handlers_catch_up.go | 233 ++++++++ internal/server/handlers_dashboard.go | 62 ++- internal/server/server.go | 6 + internal/server/task3615_catch_up_test.go | 484 ++++++++++++++++ .../server/testdata/compress_get_routes.txt | 1 + internal/store/catch_up.go | 526 ++++++++++++++++++ .../migrations/149_workspace_read_cursors.sql | 23 + internal/store/nulcolumns.go | 3 + .../119_workspace_read_cursors.sql | 12 + internal/store/read_cursors.go | 46 ++ 12 files changed, 1376 insertions(+), 24 deletions(-) create mode 100644 internal/server/handlers_catch_up.go create mode 100644 internal/server/task3615_catch_up_test.go create mode 100644 internal/store/catch_up.go create mode 100644 internal/store/migrations/149_workspace_read_cursors.sql create mode 100644 internal/store/pgmigrations/119_workspace_read_cursors.sql create mode 100644 internal/store/read_cursors.go diff --git a/docs/api.md b/docs/api.md index 805a9cb96..77cc973b8 100644 --- a/docs/api.md +++ b/docs/api.md @@ -73,6 +73,8 @@ REST API at `/api/v1/`. Key endpoints: - `POST /workspaces/{ws}/playbooks/{ref}/run` — bind a playbook's declared arguments and return its body (`pad playbook run`, `pad_playbook.action: run`). Side-effect-free: nothing is stored. So it is the ONE POST a read-scoped token (`read` / `pad:read`, PAT or OAuth) may make (TASK-2863). That is an exact-path allowlist in `tokenScopeAllows`; both variable segments must be plain names (RFC 3986 unreserved characters). `/playbooks/match` stays refused for read tokens, as do every write and the item `claim` / `release` routes - `POST /workspaces/{ws}/playbooks/match {text}` — which ACTIVE playbook (or `none`) free text asks for, answered by the decision provider synchronously (TASK-3120; `pad playbook match -- ""`, `pad_playbook.action: match`). 404 `decision_provider_unavailable` with no provider, so callers fall back to slug/trigger routing. Each call that reaches the provider spends from its own bucket, 30/min per user, burst 5, charged immediately before the provider call, so an unconfigured 404, a 400 for the text, a too-many-playbooks 400 or a zero-playbook answer consumes nothing (a request-too-large 400 is raised inside that call and has spent its token); over it is `429` with `Retry-After` (TASK-3141). No env knob by design: tuning belongs in the instance-admin decision-provider setting - `GET /workspaces/{ws}/dashboard` — computed project overview (active items, plans, attention, blockers). Also carries `pending_reminders`: fired-but-unacknowledged reminders, which is the delivery path on any instance with no webhook configured. With a decision provider, `attention` also carries `needs_human` entries (the item's latest `needs_human_decision` answer is >= 0.7), and a text-derived `blocked` entry for an item whose `blocked` answer is >= 0.7 and which the dependency graph has NOT already flagged — the graph stays primary. Only CURRENT answers count, the same rule the decisions endpoint applies (the item's present state, the question as registered now, the model pinned now), so an item edited or commented since its last evaluation drops off until the tick re-asks. The state check costs one item-plus-trail read per ABOVE-threshold item, never per workspace item. A failing provider (any owed `attention` job with a recorded error) marks the section `attention.decisions` in `degraded_sections`. Text-derived `blocked` entries therefore also reach `pad project stale` (which filters on type); `needs_human` does not +- `GET /workspaces/{ws}/catch-up` — the caller's "catch me up" digest (TASK-3615). Returns `{window: {since, until, has_cursor, capped_at_ceiling}, needs_you, finished, new_and_reopened, comments, agents, degraded?}`. Each section is `{total, lines: [{ref, title, collection, what, actor?, agent?, at?}]}` with at most 5 lines; `agents` is `{total, groups: [{agent, changes, items, lines}]}`. The window runs from `max(cursor, now - 14 days)` to now, compared at second precision with an INCLUSIVE lower bound (an event in the same second as the cursor is shown again rather than hidden). With no cursor the window is the 14-day ceiling; the GET never writes one. The caller's own actions are left out: `finished` and the reopens in `new_and_reopened` read their actor from the activity row of the same item and second, and are dropped only when every such row is the caller's (no row, or another person's, keeps the line). Agent work is grouped by agent name and keeps the caller's own agents. `needs_you` is current state: attention-set `needs_human` items, then open, non-reference items assigned to the caller that a visible, unfinished item blocks; a failing decision provider returns `degraded: ["needs_you"]`. Every line and total is limited to what the caller may see (collection access with the app-token ceiling, and item grants). 401 when signed out +- `PUT /workspaces/{ws}/me/seen` — mark the workspace caught up: sets the caller's cursor to now and returns `{seen_at}`. No body (a client-supplied time could hide unseen work), and the cursor never moves backwards. Per user and workspace; any member may call it - `GET /workspaces/{ws}/report` — the windowed Insights report: `window` (day | week | 2wk | month), `offset` (periods back), `collections` (slugs), `include_items`, and since TASK-3524 `tz` (an IANA zone such as `America/Los_Angeles`). With `tz` the throughput buckets are the viewer's local days (local hours in the day window), keyed in that zone, and the response echoes `tz`; without it they are UTC, byte for byte as before, which is what the CLI and MCP get. An unknown zone, or `Local` (the server's zone), answers 400 `invalid_tz`. The binary embeds the zone database, so this works on hosts without one. Nothing caches the response; a cache added later must key on `tz` with the other parameters and the caller's visibility scope - `GET /workspaces/{ws}/activity` — workspace activity feed (enriched with item titles + change details). Filters `action`, `actor` (`user` | `agent`), `source`, `since` (YYYY-MM-DD) and, since TASK-2219, `collection` (a collection slug: activity on items currently in that collection). All are applied in the store query before `limit`, so every page is full and the `before`/`before_id` keyset cursor walks the filtered feed. A `collection` that names no live collection, or one the caller cannot see, answers `[]`, not 404, so it is not an existence oracle - `GET/POST/DELETE /workspaces/{ws}/webhooks` — webhook management diff --git a/internal/models/plan3535_lifecycle_sites_guard_test.go b/internal/models/plan3535_lifecycle_sites_guard_test.go index 5459a2bca..13c9a6a3a 100644 --- a/internal/models/plan3535_lifecycle_sites_guard_test.go +++ b/internal/models/plan3535_lifecycle_sites_guard_test.go @@ -61,6 +61,8 @@ var lifecycleSites = map[string]string{ // Dashboard open-work sections. "internal/server/handlers_item_context.go::handleGetItemContext::isItemDone": "exclude: item-context conflicts (PLAN-3598) mirror the dashboard: the subject item checks isReferenceCollection first; a blocker keeps a reference BLOCKER, an explicit link; a closed parent of any collection is an anomaly (the close guard skips only reference children)", "internal/server/handlers_dashboard.go::buildDashboardResponse::isItemDone": "exclude: overdue, blocked, suggested_next and by_role read workItems; orphaned_task checks isReferenceCollection; the blocker check keeps a reference BLOCKER, an explicit link", + "internal/server/handlers_dashboard.go::firstActiveBlockers::isItemDone": "keep: the done check on a BLOCKER, moved out of buildDashboardResponse (TASK-3615); a reference blocker is an explicit link and still blocks", + "internal/server/handlers_catch_up.go::catchUpNeedsYou::isItemDone": "exclude: needs_you's blocked items skip isReferenceCollection first, as the dashboard's blocked attention reads workItems (TASK-3615)", // Close guard. "internal/server/handlers_items_open_children_guard.go::runOpenChildrenGuard::isItemDone": "exclude: a reference child is skipped before the done check", diff --git a/internal/server/handlers_catch_up.go b/internal/server/handlers_catch_up.go new file mode 100644 index 000000000..f89b90119 --- /dev/null +++ b/internal/server/handlers_catch_up.go @@ -0,0 +1,233 @@ +package server + +import ( + "log/slog" + "net/http" + "sort" + "time" + + "github.com/PerpetualSoftware/pad/internal/decision" + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// catchUpCeiling is how far back the catch-up digest ever reaches +// (TASK-3615, Dave's 10-17 answer). With no cursor the window IS the ceiling. +const catchUpCeiling = 14 * 24 * time.Hour + +// catchUpWindow is the period the digest covers. +type catchUpWindow struct { + Since string `json:"since"` + Until string `json:"until"` + // HasCursor is false until the viewer first marks the workspace caught up. + HasCursor bool `json:"has_cursor"` + // CappedAtCeiling says the window starts at the 14-day ceiling rather + // than at the cursor: there is no cursor, or it is older than that. + CappedAtCeiling bool `json:"capped_at_ceiling"` +} + +type catchUpResponse struct { + Window catchUpWindow `json:"window"` + NeedsYou store.CatchUpSection `json:"needs_you"` + Finished store.CatchUpSection `json:"finished"` + NewAndReopened store.CatchUpSection `json:"new_and_reopened"` + Comments store.CatchUpSection `json:"comments"` + Agents store.CatchUpAgents `json:"agents"` + // Degraded names a section whose source failed and came back empty, so a + // client can tell "nothing" from "could not read" (the dashboard's rule, + // BUG-2014). Today only needs_you's decision source can degrade. + Degraded []string `json:"degraded,omitempty"` +} + +// handleGetCatchUp returns the per-user "catch me up" digest (TASK-3615): +// what others did in the workspace since the viewer last marked it caught +// up, at most 14 days back, plus what needs the viewer now. A GET never +// writes: with no cursor the window is computed, not stored, and only +// PUT /me/seen creates or moves the cursor. +// +// GET /workspaces/{slug}/catch-up +func (s *Server) handleGetCatchUp(w http.ResponseWriter, r *http.Request) { + workspaceID, ok := s.getWorkspaceID(w, r) + if !ok { + return + } + user := currentUser(r) + if user == nil { + writeError(w, http.StatusUnauthorized, "unauthorized", "Sign in to see what changed since your last visit") + return + } + + until := time.Now().UTC() + ceiling := until.Add(-catchUpCeiling) + cursor, found, err := s.store.ReadCursor(user.ID, workspaceID) + if err != nil { + writeInternalError(w, err) + return + } + since := ceiling + if found && cursor.After(ceiling) { + since = cursor + } + + // The same visibility every list door applies: collection access (with + // an app token's ceiling), and item grants for a guest or a restricted + // member. Totals are counted under the same scope as the lines, so a + // total cannot reveal an item the viewer cannot open. + visibleIDs, err := s.visibleCollectionIDs(r, workspaceID) + if err != nil { + writeInternalError(w, err) + return + } + fullCollIDs, grantedItemIDs, err := s.guestResourceFilter(r, workspaceID) + if err != nil { + writeInternalError(w, err) + return + } + scope := store.CatchUpScope{CollectionIDs: visibleIDs} + if len(grantedItemIDs) > 0 { + scope = store.CatchUpScope{CollectionIDs: fullCollIDs, ItemIDs: grantedItemIDs} + } + // A nil CollectionIDs means "no restriction" to the store, so a grant + // holder with no full-collection access must carry an empty, non-nil set. + if scope.CollectionIDs == nil && len(grantedItemIDs) > 0 { + scope.CollectionIDs = []string{} + } + + cu, err := s.store.CatchUp(workspaceID, user.ID, since, until, scope) + if err != nil { + writeInternalError(w, err) + return + } + + resp := catchUpResponse{ + Window: catchUpWindow{ + Since: since.Format(time.RFC3339), + Until: until.Format(time.RFC3339), + HasCursor: found, + CappedAtCeiling: !since.After(ceiling), + }, + Finished: cu.Finished, + NewAndReopened: cu.NewAndReopened, + Comments: cu.Comments, + Agents: cu.Agents, + } + + needs, degraded, err := s.catchUpNeedsYou(r, workspaceID, user.ID, visibleIDs, fullCollIDs, grantedItemIDs, scope) + if err != nil { + writeInternalError(w, err) + return + } + resp.NeedsYou = needs + if degraded { + resp.Degraded = []string{"needs_you"} + } + writeJSON(w, http.StatusOK, resp) +} + +// catchUpNeedsYou is current state, not windowed: items the attention set +// says are waiting on a human, then open items assigned to the viewer that +// something visible and unfinished still blocks. Roles are agent roles, so +// "yours" means assigned to you (lead ruling, v1). A failing decision +// provider degrades the section rather than failing the digest. +func (s *Server) catchUpNeedsYou(r *http.Request, workspaceID, viewerID string, visibleIDs, fullCollIDs, grantedItemIDs []string, scope store.CatchUpScope) (store.CatchUpSection, bool, error) { + var ids []string + what := map[string]string{} + degraded := false + + flags, failing, err := s.decisionRunner().WorkspaceFlags(workspaceID, decision.AttentionSetName, decision.AttentionThreshold) + if err != nil { + slog.Error("catch-up: attention decisions failed; needs_you degraded", "workspace_id", workspaceID, "error", err) + degraded = true + } else { + if failing { + slog.Warn("catch-up: decision provider is failing; attention answers may be out of date", "workspace_id", workspaceID) + degraded = true + } + type flagged struct { + id string + p float64 + } + var fl []flagged + for id, answers := range flags { + if p, ok := answers[decision.AttentionNeedsHuman]; ok { + fl = append(fl, flagged{id, p}) + } + } + sort.Slice(fl, func(a, b int) bool { + if fl[a].p != fl[b].p { + return fl[a].p > fl[b].p + } + return fl[a].id < fl[b].id + }) + for _, f := range fl { + ids = append(ids, f.id) + what[f.id] = "needs a decision" + } + } + + if !(scope.CollectionIDs != nil && len(scope.CollectionIDs) == 0 && len(scope.ItemIDs) == 0) { + collections, err := s.store.ListCollectionsMinimal(workspaceID) + if err != nil { + return store.CatchUpSection{}, false, err + } + ctxMap := buildDoneContextMap(collections) + blockers, err := s.firstActiveBlockers(r, workspaceID, visibleIDs, fullCollIDs, grantedItemIDs, ctxMap) + if err != nil { + return store.CatchUpSection{}, false, err + } + if len(blockers) > 0 { + mine, err := s.store.ListItems(workspaceID, models.ItemListParams{ + AssignedUserID: viewerID, + CollectionIDs: scope.CollectionIDs, + ItemIDs: scope.ItemIDs, + }) + if err != nil { + return store.CatchUpSection{}, false, err + } + for _, it := range mine { + if _, blocked := blockers[it.ID]; !blocked { + continue + } + // Reference items (docs) are not open work (PLAN-3535), as + // the dashboard's blocked attention, which reads workItems. + if isReferenceCollection(it.CollectionID, ctxMap) || isItemDone(it.Fields, it.CollectionID, ctxMap) { + continue + } + if _, dup := what[it.ID]; dup { + continue + } + ids = append(ids, it.ID) + what[it.ID] = "blocked" + } + } + } + + sec, err := s.store.CatchUpItemLines(workspaceID, ids, what, scope) + if err != nil { + return store.CatchUpSection{}, false, err + } + return sec, degraded, nil +} + +// handleMarkCaughtUp moves the viewer's cursor for the workspace to now. It +// takes no body: a client-supplied time could move the cursor into the +// future and hide what has not been seen. The cursor never moves backwards. +// +// PUT /workspaces/{slug}/me/seen +func (s *Server) handleMarkCaughtUp(w http.ResponseWriter, r *http.Request) { + workspaceID, ok := s.getWorkspaceID(w, r) + if !ok { + return + } + user := currentUser(r) + if user == nil { + writeError(w, http.StatusUnauthorized, "unauthorized", "Sign in to mark this workspace caught up") + return + } + seen, err := s.store.MarkCaughtUp(user.ID, workspaceID, time.Now()) + if err != nil { + writeInternalError(w, err) + return + } + writeJSON(w, http.StatusOK, map[string]string{"seen_at": seen.UTC().Format(time.RFC3339)}) +} diff --git a/internal/server/handlers_dashboard.go b/internal/server/handlers_dashboard.go index 7a00d12fd..57b713ab5 100644 --- a/internal/server/handlers_dashboard.go +++ b/internal/server/handlers_dashboard.go @@ -269,6 +269,43 @@ type activeBlocker struct { status string } +// firstActiveBlockers resolves, for every blocked target in the workspace, +// the FIRST blocker (newest link first) that the caller can see AND that is +// still non-done. Presence in the map means the target has an active +// blocker. One GetBlocksEdges query (BUG-2002). Shared by the dashboard's +// blocked attention and the catch-up digest's needs_you (TASK-3615), so the +// two can never disagree about what blocks an item. +func (s *Server) firstActiveBlockers(r *http.Request, workspaceID string, visibleIDs, fullCollIDs, grantedItemIDs []string, ctxMap map[string]doneContext) (map[string]activeBlocker, error) { + blocksEdges, err := s.store.GetBlocksEdges(workspaceID) + if err != nil { + return nil, err + } + firstActiveBlocker := make(map[string]activeBlocker) + for _, e := range blocksEdges { + if _, resolved := firstActiveBlocker[e.TargetID]; resolved { + continue // already picked this target's first active blocker + } + // Skip blockers from hidden collections or ungrantable items — a + // blocker the caller can't see must not influence the target. + if !isCollectionVisible(e.SourceCollectionID, visibleIDs) { + continue + } + if !s.isItemVisibleToGuest(r, workspaceID, &models.Item{ID: e.SourceID, CollectionID: e.SourceCollectionID}, fullCollIDs, grantedItemIDs) { + continue + } + if isItemDone(e.SourceFields, e.SourceCollectionID, ctxMap) { + continue + } + firstActiveBlocker[e.TargetID] = activeBlocker{ + title: e.SourceTitle, + // Used only in the human-readable "Reason" string — the terminal + // check above owns the actual done-detection. + status: extractFieldValue(e.SourceFields, "status"), + } + } + return firstActiveBlocker, nil +} + // buildDoneContextMap builds a map of collection ID → (schema, settings) // for quick lookups during dashboard / lineage traversals. func buildDoneContextMap(collections []models.Collection) map[string]doneContext { @@ -837,33 +874,10 @@ func (s *Server) buildDashboardResponse(workspaceID string, r *http.Request) (*D // dashboard latency (BUG-2002). Presence in the map == the target has an // active blocker, which also drives the suggested_next blocked-filter // below. - blocksEdges, err := s.store.GetBlocksEdges(workspaceID) + firstActiveBlocker, err := s.firstActiveBlockers(r, workspaceID, visibleIDs, dashFullCollIDs, dashGrantedItemIDs, ctxMap) if err != nil { return nil, err } - firstActiveBlocker := make(map[string]activeBlocker) - for _, e := range blocksEdges { - if _, resolved := firstActiveBlocker[e.TargetID]; resolved { - continue // already picked this target's first active blocker - } - // Skip blockers from hidden collections or ungrantable items — a - // blocker the caller can't see must not influence the target. - if !isCollectionVisible(e.SourceCollectionID, visibleIDs) { - continue - } - if !s.isItemVisibleToGuest(r, workspaceID, &models.Item{ID: e.SourceID, CollectionID: e.SourceCollectionID}, dashFullCollIDs, dashGrantedItemIDs) { - continue - } - if isItemDone(e.SourceFields, e.SourceCollectionID, ctxMap) { - continue - } - firstActiveBlocker[e.TargetID] = activeBlocker{ - title: e.SourceTitle, - // Used only in the human-readable "Reason" string — the terminal - // check above owns the actual done-detection. - status: extractFieldValue(e.SourceFields, "status"), - } - } for _, item := range workItems { if isItemDone(item.Fields, item.CollectionID, ctxMap) { continue diff --git a/internal/server/server.go b/internal/server/server.go index b71a746ad..38416d156 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -2487,6 +2487,12 @@ func (s *Server) setupRouter() { r.Get("/report/layout", s.handleGetReportLayout) r.Put("/report/layout", s.handleSaveReportLayout) + // Per-user "catch me up" digest and its cursor + // (TASK-3615). The GET never writes; only PUT + // /me/seen creates or moves the cursor. + r.Get("/catch-up", s.handleGetCatchUp) + r.Put("/me/seen", s.handleMarkCaughtUp) + // Project intelligence reads — next/standup/changelog // (PLAN-1888 / TASK-1894). Mirror `pad project // next|standup|changelog` (cmd/pad/main.go) — KEEP IN diff --git a/internal/server/task3615_catch_up_test.go b/internal/server/task3615_catch_up_test.go new file mode 100644 index 000000000..1a8a31b95 --- /dev/null +++ b/internal/server/task3615_catch_up_test.go @@ -0,0 +1,484 @@ +package server + +import ( + "fmt" + "net/http" + "sort" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// TASK-3615: the "catch me up" digest. Every fixture acts through the HTTP +// doors as real users, so the activity rows carry the user ids the actor +// join reads, then pins every timestamp in the workspace to one second +// (task3615At). The join is "same item, same second" by ruling, so a run +// that straddled a second boundary between a transition and its activity row +// would otherwise flake; pinning keeps the test about the rule, not the +// clock. + +type task3615Fixture struct { + srv *Server + ws *models.Workspace + alice, bob *models.User + aliceTok string + bobTok string + tasksID string + ideasID string + collectionID map[string]string +} + +func newTask3615Fixture(t *testing.T, srv *Server) *task3615Fixture { + t.Helper() + slug := createWSWithCollections(t, srv) + ws, err := srv.store.GetWorkspaceBySlug(slug) + if err != nil || ws == nil { + t.Fatalf("GetWorkspaceBySlug: %v", err) + } + f := &task3615Fixture{srv: srv, ws: ws, collectionID: map[string]string{}} + f.alice = f.user(t, "alice", "owner") + f.bob = f.user(t, "bob", "editor") + f.aliceTok = f.session(t, f.alice) + f.bobTok = f.session(t, f.bob) + for _, c := range []string{"tasks", "ideas"} { + coll, err := srv.store.GetCollectionBySlug(ws.ID, c) + if err != nil || coll == nil { + t.Fatalf("GetCollectionBySlug %s: %v", c, err) + } + f.collectionID[c] = coll.ID + } + f.tasksID, f.ideasID = f.collectionID["tasks"], f.collectionID["ideas"] + return f +} + +func (f *task3615Fixture) user(t *testing.T, name, role string) *models.User { + t.Helper() + u, err := f.srv.store.CreateUser(models.UserCreate{Email: name + "-3615@example.com", Name: name, Username: name + "-3615", Password: "pw-test-12345"}) + if err != nil { + t.Fatalf("CreateUser %s: %v", name, err) + } + if err := f.srv.store.AddWorkspaceMember(f.ws.ID, u.ID, role); err != nil { + t.Fatalf("AddWorkspaceMember %s: %v", name, err) + } + return u +} + +func (f *task3615Fixture) session(t *testing.T, u *models.User) string { + t.Helper() + tok, err := f.srv.store.CreateSession(u.ID, "go-test", "192.0.2.1", "go-test", 24*time.Hour) + if err != nil { + t.Fatalf("CreateSession: %v", err) + } + return tok +} + +func (f *task3615Fixture) create(t *testing.T, tok, coll, title string) models.Item { + t.Helper() + rr := doRequestWithCookie(f.srv, "POST", "/api/v1/workspaces/"+f.ws.Slug+"/collections/"+coll+"/items", map[string]any{"title": title}, tok) + if rr.Code != http.StatusCreated { + t.Fatalf("create %q: %d %s", title, rr.Code, rr.Body.String()) + } + var it models.Item + parseJSON(t, rr, &it) + return it +} + +func (f *task3615Fixture) patch(t *testing.T, tok string, it models.Item, body map[string]any) { + t.Helper() + rr := doRequestWithCookie(f.srv, "PATCH", "/api/v1/workspaces/"+f.ws.Slug+"/items/"+it.Slug, body, tok) + if rr.Code != http.StatusOK { + t.Fatalf("patch %s %v: %d %s", it.Title, body, rr.Code, rr.Body.String()) + } +} + +func (f *task3615Fixture) status(t *testing.T, tok string, it models.Item, status string) { + t.Helper() + f.patch(t, tok, it, map[string]any{"fields_patch": map[string]any{"status": status}}) +} + +func (f *task3615Fixture) comment(t *testing.T, tok string, it models.Item, body string) { + t.Helper() + rr := doRequestWithCookie(f.srv, "POST", "/api/v1/workspaces/"+f.ws.Slug+"/items/"+it.Slug+"/comments", map[string]any{"body": body}, tok) + if rr.Code != http.StatusCreated { + t.Fatalf("comment on %s: %d %s", it.Title, rr.Code, rr.Body.String()) + } +} + +// pin sets every timestamp the digest reads in this workspace to at. +func (f *task3615Fixture) pin(t *testing.T, at time.Time) { + t.Helper() + ts := at.UTC().Format(time.RFC3339) + for _, tbl := range []string{"items", "status_transitions", "activities", "comments"} { + if _, err := f.srv.store.DB().Exec(fmt.Sprintf("UPDATE %s SET created_at = '%s' WHERE workspace_id = '%s'", tbl, ts, f.ws.ID)); err != nil { + t.Fatalf("pin %s: %v", tbl, err) + } + } +} + +func (f *task3615Fixture) catchUp(t *testing.T, tok string) catchUpResponse { + t.Helper() + rr := doRequestWithCookie(f.srv, "GET", "/api/v1/workspaces/"+f.ws.Slug+"/catch-up", nil, tok) + if rr.Code != http.StatusOK { + t.Fatalf("GET catch-up: %d %s", rr.Code, rr.Body.String()) + } + var resp catchUpResponse + parseJSON(t, rr, &resp) + return resp +} + +func task3615Titles(sec store.CatchUpSection) []string { + out := []string{} + for _, l := range sec.Lines { + out = append(out, l.Title) + } + sort.Strings(out) + return out +} + +func task3615Want(t *testing.T, name string, sec store.CatchUpSection, want ...string) { + t.Helper() + sort.Strings(want) + got := task3615Titles(sec) + if fmt.Sprint(got) != fmt.Sprint(want) || sec.Total != len(want) { + t.Errorf("%s: lines %v total %d, want %v total %d", name, got, sec.Total, want, len(want)) + } +} + +var task3615At = time.Now().UTC().Add(-time.Hour).Truncate(time.Second) + +func TestTASK3615_CatchUpSections(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + + mine := f.create(t, f.aliceTok, "tasks", "Alice's task") + f.comment(t, f.bobTok, mine, "bob on alice's task") + f.comment(t, f.aliceTok, mine, "alice on her own task") + + bobDone := f.create(t, f.bobTok, "tasks", "Bob finished") + f.status(t, f.bobTok, bobDone, "done") + + aliceDone := f.create(t, f.aliceTok, "tasks", "Alice finished") + f.status(t, f.aliceTok, aliceDone, "done") + + // Alice finished it, but Bob wrote to it in the same second: the + // actor is ambiguous, so the line stays. + shared := f.create(t, f.aliceTok, "tasks", "Shared second") + f.status(t, f.aliceTok, shared, "done") + f.patch(t, f.bobTok, shared, map[string]any{"title": "Shared second (bob)"}) + + // Bob's alone: pinning puts every row of an item in one second, so + // a reopen by someone other than the finisher would make the finish + // ambiguous too. + reopened := f.create(t, f.bobTok, "tasks", "Bob reopened") + f.status(t, f.bobTok, reopened, "done") + f.status(t, f.bobTok, reopened, "open") + + f.create(t, f.bobTok, "ideas", "Bob's idea") + + // Alice's own item, reopened by Bob: news as a reopen, not as a + // creation (only Bob's rows touch its status). + bobReopensMine := f.create(t, f.aliceTok, "tasks", "Bob reopened mine") + f.status(t, f.bobTok, bobReopensMine, "done") + f.status(t, f.bobTok, bobReopensMine, "open") + + // Cancelled is terminal but abandoned: not a finish. + cancelled := f.create(t, f.aliceTok, "tasks", "Bob cancelled") + f.status(t, f.bobTok, cancelled, "cancelled") + + // Comments that are not Alice's news: her own, alone, on her item; + // and Bob's on his own item, which is neither hers nor assigned to her. + ownComment := f.create(t, f.aliceTok, "tasks", "Only Alice commented") + f.comment(t, f.aliceTok, ownComment, "note to self") + bobsOwn := f.create(t, f.bobTok, "tasks", "Bob's own thread") + f.comment(t, f.bobTok, bobsOwn, "bob talking to himself") + + f.pin(t, task3615At) + if _, err := srv.store.MarkCaughtUp(f.alice.ID, f.ws.ID, task3615At.Add(-time.Hour)); err != nil { + t.Fatalf("MarkCaughtUp: %v", err) + } + + got := f.catchUp(t, f.aliceTok) + task3615Want(t, "finished", got.Finished, "Bob finished", "Bob reopened", "Bob reopened mine", "Shared second (bob)") + task3615Want(t, "new_and_reopened", got.NewAndReopened, "Bob finished", "Bob's idea", "Bob reopened", "Bob reopened mine", "Bob's own thread") + task3615Want(t, "comments", got.Comments, "Alice's task") + for _, l := range got.NewAndReopened.Lines { + if l.Title == "Bob reopened mine" && l.What != "reopened" { + t.Errorf("%q: what %q, want reopened", l.Title, l.What) + } + } + if !got.Window.HasCursor || got.Window.CappedAtCeiling { + t.Errorf("window %+v: want a cursor inside the ceiling", got.Window) + } + for _, l := range got.Finished.Lines { + if l.Actor != "bob" { + t.Errorf("finished %q: actor %q, want bob", l.Title, l.Actor) + } + } + + // Bob's view is the mirror: Alice's finishes are his news, his own + // are not. + bobView := f.catchUp(t, f.bobTok) + task3615Want(t, "bob finished", bobView.Finished, "Alice finished", "Shared second (bob)") + }) +} + +func TestTASK3615_CursorAndCeilingBoundTheWindow(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + it := f.create(t, f.bobTok, "tasks", "Bob finished") + f.status(t, f.bobTok, it, "done") + f.pin(t, task3615At) + + // No cursor: the window is the 14-day ceiling, and reading it does + // not create one. + got := f.catchUp(t, f.aliceTok) + task3615Want(t, "no cursor", got.Finished, "Bob finished") + if got.Window.HasCursor || !got.Window.CappedAtCeiling { + t.Errorf("window %+v: want no cursor, capped at the ceiling", got.Window) + } + if _, found, err := srv.store.ReadCursor(f.alice.ID, f.ws.ID); err != nil || found { + t.Fatalf("GET wrote a cursor (found=%v, err=%v)", found, err) + } + + // A press in the same second as the event does NOT hide it: the + // event is stamped to the second and may have come after the press. + if _, err := srv.store.MarkCaughtUp(f.alice.ID, f.ws.ID, task3615At.Add(500*time.Millisecond)); err != nil { + t.Fatalf("MarkCaughtUp: %v", err) + } + task3615Want(t, "same-second cursor", f.catchUp(t, f.aliceTok).Finished, "Bob finished") + // A press in the next second does. + if _, err := srv.store.MarkCaughtUp(f.alice.ID, f.ws.ID, task3615At.Add(time.Second)); err != nil { + t.Fatalf("MarkCaughtUp: %v", err) + } + task3615Want(t, "next-second cursor", f.catchUp(t, f.aliceTok).Finished) + + // The ceiling: an event 15 days old is out even with no cursor. + bobView := f.catchUp(t, f.bobTok) + task3615Want(t, "bob, own work", bobView.Finished) + f.pin(t, time.Now().UTC().Add(-15*24*time.Hour)) + other := f.user(t, "carol", "editor") + task3615Want(t, "past the ceiling", f.catchUp(t, f.session(t, other)).Finished) + + // A cursor older than the ceiling does not reach past it. + dave := f.user(t, "dave", "editor") + if _, err := srv.store.MarkCaughtUp(dave.ID, f.ws.ID, time.Now().UTC().Add(-20*24*time.Hour)); err != nil { + t.Fatalf("MarkCaughtUp: %v", err) + } + old := f.catchUp(t, f.session(t, dave)) + task3615Want(t, "old cursor, past the ceiling", old.Finished) + if !old.Window.HasCursor || !old.Window.CappedAtCeiling { + t.Errorf("old cursor window %+v: want a cursor, capped at the ceiling", old.Window) + } + }) +} + +func TestTASK3615_MarkCaughtUpNeverMovesBackwards(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + rr := doRequestWithCookie(srv, "PUT", "/api/v1/workspaces/"+f.ws.Slug+"/me/seen", nil, f.aliceTok) + if rr.Code != http.StatusOK { + t.Fatalf("PUT me/seen: %d %s", rr.Code, rr.Body.String()) + } + first, found, err := srv.store.ReadCursor(f.alice.ID, f.ws.ID) + if err != nil || !found { + t.Fatalf("cursor after PUT: found=%v err=%v", found, err) + } + if d := time.Since(first); d < 0 || d > time.Minute { + t.Fatalf("cursor %v is not now", first) + } + back, err := srv.store.MarkCaughtUp(f.alice.ID, f.ws.ID, first.Add(-time.Hour)) + if err != nil { + t.Fatalf("MarkCaughtUp: %v", err) + } + if !back.Equal(first) { + t.Fatalf("cursor moved backwards: %v, was %v", back, first) + } + // Per user: Bob has no cursor. + if _, found, _ := srv.store.ReadCursor(f.bob.ID, f.ws.ID); found { + t.Fatalf("Alice's press created Bob's cursor") + } + }) +} + +func TestTASK3615_VisibilityBoundsLinesAndTotals(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + task := f.create(t, f.bobTok, "tasks", "Visible task") + granted := f.create(t, f.bobTok, "ideas", "Granted idea") + hidden := f.create(t, f.bobTok, "ideas", "Hidden idea") + f.status(t, f.bobTok, task, "done") + f.status(t, f.bobTok, granted, "implemented") + f.status(t, f.bobTok, hidden, "implemented") + f.pin(t, task3615At) + + restricted := f.user(t, "restricted", "editor") + if err := srv.store.SetMemberCollectionAccess(f.ws.ID, restricted.ID, "specific", []string{f.tasksID}); err != nil { + t.Fatalf("SetMemberCollectionAccess: %v", err) + } + tok := f.session(t, restricted) + + // Control: the owner sees all three, so the fixture is what it says. + task3615Want(t, "owner", f.catchUp(t, f.aliceTok).Finished, "Visible task", "Granted idea", "Hidden idea") + + got := f.catchUp(t, tok) + task3615Want(t, "restricted finished", got.Finished, "Visible task") + task3615Want(t, "restricted new", got.NewAndReopened, "Visible task") + + if _, err := srv.store.CreateItemGrant(f.ws.ID, granted.ID, restricted.ID, "view", f.alice.ID); err != nil { + t.Fatalf("CreateItemGrant: %v", err) + } + got = f.catchUp(t, tok) + task3615Want(t, "granted finished", got.Finished, "Visible task", "Granted idea") + task3615Want(t, "granted new", got.NewAndReopened, "Visible task", "Granted idea") + }) +} + +func TestTASK3615_NeedsYouListsBlockedItemsAssignedToYou(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + blocked := f.create(t, f.bobTok, "tasks", "Assigned and blocked") + f.patch(t, f.bobTok, blocked, map[string]any{"assigned_user_id": f.alice.ID}) + notMine := f.create(t, f.bobTok, "tasks", "Blocked, not mine") + freed := f.create(t, f.bobTok, "tasks", "Assigned, blocker done") + f.patch(t, f.bobTok, freed, map[string]any{"assigned_user_id": f.alice.ID}) + doneButBlocked := f.create(t, f.bobTok, "tasks", "Assigned, done, link left") + f.patch(t, f.bobTok, doneButBlocked, map[string]any{"assigned_user_id": f.alice.ID}) + f.status(t, f.bobTok, doneButBlocked, "done") + // A doc is reference, not open work (PLAN-3535): assigned and + // blocked, it is still not something that needs Alice. + refDoc := f.create(t, f.bobTok, "docs", "Assigned blocked doc") + f.patch(t, f.bobTok, refDoc, map[string]any{"assigned_user_id": f.alice.ID}) + blocker := f.create(t, f.bobTok, "tasks", "Blocker") + doneBlocker := f.create(t, f.bobTok, "tasks", "Done blocker") + f.status(t, f.bobTok, doneBlocker, "done") + for _, l := range []struct{ src, tgt models.Item }{{blocker, blocked}, {blocker, notMine}, {doneBlocker, freed}, {blocker, doneButBlocked}, {blocker, refDoc}} { + if _, err := srv.store.CreateItemLink(f.ws.ID, models.ItemLinkCreate{TargetID: l.tgt.ID, LinkType: "blocks"}, l.src.ID); err != nil { + t.Fatalf("CreateItemLink: %v", err) + } + } + got := f.catchUp(t, f.aliceTok) + task3615Want(t, "needs_you", got.NeedsYou, "Assigned and blocked") + if len(got.NeedsYou.Lines) == 1 && got.NeedsYou.Lines[0].What != "blocked" { + t.Errorf("needs_you what = %q, want blocked", got.NeedsYou.Lines[0].What) + } + }) +} + +func TestTASK3615_AgentsGroupedByName(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + a := f.create(t, f.bobTok, "tasks", "Agent target A") + b := f.create(t, f.bobTok, "tasks", "Agent target B") + seed := func(doc models.Item, meta string) { + if _, err := srv.store.CreateActivity(models.Activity{WorkspaceID: f.ws.ID, DocumentID: doc.ID, Action: "updated", Actor: "agent", Source: "cli", Metadata: meta, UserID: f.alice.ID}); err != nil { + t.Fatalf("CreateActivity: %v", err) + } + } + seed(a, `{"agent":"wren"}`) + seed(a, `{"agent":"wren"}`) + seed(b, `{"agent":"wren"}`) + seed(b, `{"agent":"rook"}`) + // Malformed metadata can only exist on SQLite: Postgres's JSONB + // refuses it at write time. + malformed := srv.store.D().Driver() != store.DriverPostgres + if malformed { + seed(b, `not json`) + } + f.pin(t, task3615At) + + got := f.catchUp(t, f.aliceTok).Agents + byName := map[string]store.CatchUpAgentGroup{} + for _, g := range got.Groups { + byName[g.Agent] = g + } + // The viewer's own agents stay in: these rows carry Alice's user id. + if g := byName["wren"]; g.Changes != 3 || g.Items != 2 { + t.Errorf("wren: %+v, want 3 changes on 2 items", g) + } + if g := byName["rook"]; g.Changes != 1 || g.Items != 1 { + t.Errorf("rook: %+v, want 1 change on 1 item", g) + } + wantTotal := 2 + if malformed { + wantTotal = 3 + if g := byName["agent"]; g.Changes != 1 { + t.Errorf("malformed metadata: %+v, want one unnamed change", g) + } + } + if got.Total != wantTotal { + t.Errorf("agents total %d, want %d", got.Total, wantTotal) + } + }) +} + +func TestTASK3615_SignedOutIsRefused(t *testing.T) { + t.Parallel() + srv := testServer(t) + f := newTask3615Fixture(t, srv) + for _, m := range []string{"GET /catch-up", "PUT /me/seen"} { + method, path := m[:3], m[4:] + rr := doRequest(srv, method, "/api/v1/workspaces/"+f.ws.Slug+path, nil) + // A signed-out write is refused by middleware (403) before the + // handler's own 401; either way it is refused. + if rr.Code != http.StatusUnauthorized && rr.Code != http.StatusForbidden { + t.Errorf("%s signed out: %d, want 401 or 403", m, rr.Code) + } + } +} + +// CatchUpItemLines is the only visibility gate on needs_you's attention-set +// ids (the decision runner's map is workspace-wide), so it is pinned here +// directly rather than through a provider. +func TestTASK3615_NeedsYouLinesAreScoped(t *testing.T) { + t.Parallel() + bothBackends(t, func(t *testing.T, srv *Server) { + f := newTask3615Fixture(t, srv) + task := f.create(t, f.bobTok, "tasks", "Visible") + idea := f.create(t, f.bobTok, "ideas", "Hidden") + what := map[string]string{task.ID: "needs a decision", idea.ID: "needs a decision"} + ids := []string{idea.ID, task.ID} + + all, err := srv.store.CatchUpItemLines(f.ws.ID, ids, what, store.CatchUpScope{}) + if err != nil { + t.Fatalf("CatchUpItemLines: %v", err) + } + task3615Want(t, "unscoped control", all, "Visible", "Hidden") + if all.Lines[0].Title != "Hidden" { + t.Errorf("order not kept: %v", task3615TitlesInOrder(all)) + } + + scoped, err := srv.store.CatchUpItemLines(f.ws.ID, ids, what, store.CatchUpScope{CollectionIDs: []string{f.tasksID}}) + if err != nil { + t.Fatalf("CatchUpItemLines: %v", err) + } + task3615Want(t, "scoped", scoped, "Visible") + + granted, err := srv.store.CatchUpItemLines(f.ws.ID, ids, what, store.CatchUpScope{CollectionIDs: []string{}, ItemIDs: []string{idea.ID}}) + if err != nil { + t.Fatalf("CatchUpItemLines: %v", err) + } + task3615Want(t, "grant only", granted, "Hidden") + + none, err := srv.store.CatchUpItemLines(f.ws.ID, ids, what, store.CatchUpScope{CollectionIDs: []string{}}) + if err != nil { + t.Fatalf("CatchUpItemLines: %v", err) + } + task3615Want(t, "sees nothing", none) + }) +} + +func task3615TitlesInOrder(sec store.CatchUpSection) []string { + out := []string{} + for _, l := range sec.Lines { + out = append(out, l.Title) + } + return out +} diff --git a/internal/server/testdata/compress_get_routes.txt b/internal/server/testdata/compress_get_routes.txt index 7386f7af7..2a39da477 100644 --- a/internal/server/testdata/compress_get_routes.txt +++ b/internal/server/testdata/compress_get_routes.txt @@ -85,6 +85,7 @@ compress /api/v1/workspaces/{slug}/archived-collections compress /api/v1/workspaces/{slug}/attachments skip /api/v1/workspaces/{slug}/attachments/{attachmentID} compress /api/v1/workspaces/{slug}/builtins +compress /api/v1/workspaces/{slug}/catch-up compress /api/v1/workspaces/{slug}/changelog compress /api/v1/workspaces/{slug}/changes skip /api/v1/workspaces/{slug}/claim-code diff --git a/internal/store/catch_up.go b/internal/store/catch_up.go new file mode 100644 index 000000000..d7eeb6063 --- /dev/null +++ b/internal/store/catch_up.go @@ -0,0 +1,526 @@ +package store + +import ( + "fmt" + "sort" + "strings" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +// CatchUpLineCap is how many lines each catch-up section carries. The rest +// is in the section's total ("and N more"). +const CatchUpLineCap = 5 + +// CatchUpScope is the viewer's item visibility, in the shape every list door +// already computes (TASK-3615): CollectionIDs nil means no collection +// restriction (a cookie-session admin); CollectionIDs non-nil and empty with +// no ItemIDs means the viewer sees nothing. ItemIDs are item-level grants, +// additive to CollectionIDs. +type CatchUpScope struct { + CollectionIDs []string + ItemIDs []string +} + +func (sc CatchUpScope) seesNothing() bool { + return sc.CollectionIDs != nil && len(sc.CollectionIDs) == 0 && len(sc.ItemIDs) == 0 +} + +// clause renders the scope as a predicate on the items alias. +func (sc CatchUpScope) clause(alias string) (string, []any) { + ph := func(ids []string, args *[]any) string { + p := make([]string, len(ids)) + for i, id := range ids { + p[i] = "?" + *args = append(*args, id) + } + return strings.Join(p, ",") + } + var args []any + switch { + case len(sc.CollectionIDs) > 0 && len(sc.ItemIDs) > 0: + c := ph(sc.CollectionIDs, &args) + it := ph(sc.ItemIDs, &args) + return fmt.Sprintf(" AND (%s.collection_id IN (%s) OR %s.id IN (%s))", alias, c, alias, it), args + case len(sc.CollectionIDs) > 0: + return fmt.Sprintf(" AND %s.collection_id IN (%s)", alias, ph(sc.CollectionIDs, &args)), args + case len(sc.ItemIDs) > 0: + return fmt.Sprintf(" AND %s.id IN (%s)", alias, ph(sc.ItemIDs, &args)), args + } + return "", nil +} + +// CatchUpLine is one item in a catch-up section. +type CatchUpLine struct { + Ref string `json:"ref"` + Title string `json:"title"` + Collection string `json:"collection"` + // What happened, in a word or two: "finished", "created", "reopened", + // "commented", "needs a decision", "blocked", or an activity action. + What string `json:"what"` + Actor string `json:"actor,omitempty"` + Agent string `json:"agent,omitempty"` + At string `json:"at,omitempty"` + + itemID string +} + +// CatchUpSection is a capped list with the full count beside it. +type CatchUpSection struct { + Total int `json:"total"` + Lines []CatchUpLine `json:"lines"` +} + +// CatchUpAgentGroup is one agent's work in the window. +type CatchUpAgentGroup struct { + Agent string `json:"agent"` + Changes int `json:"changes"` + Items int `json:"items"` + Lines []CatchUpLine `json:"lines"` +} + +// CatchUpAgents is the agents section: one group per agent name. +type CatchUpAgents struct { + Total int `json:"total"` + Groups []CatchUpAgentGroup `json:"groups"` +} + +// CatchUp is the windowed half of the digest. needs_you is current state, +// not windowed, and is assembled by the server from the attention set. +type CatchUp struct { + Finished CatchUpSection `json:"finished"` + NewAndReopened CatchUpSection `json:"new_and_reopened"` + Comments CatchUpSection `json:"comments"` + Agents CatchUpAgents `json:"agents"` +} + +func capSection(lines []CatchUpLine) CatchUpSection { + sec := CatchUpSection{Total: len(lines), Lines: lines} + if len(sec.Lines) > CatchUpLineCap { + sec.Lines = sec.Lines[:CatchUpLineCap] + } + if sec.Lines == nil { + sec.Lines = []CatchUpLine{} + } + return sec +} + +// CatchUp reads what happened in workspaceID in (since, until] that viewerID +// did not do, limited to scope. Bounds are compared at second precision, the +// precision the source tables store, and the lower bound is INCLUSIVE: an +// event stamped in the same second as since may have happened after the +// press, so it is shown again rather than risk hiding it. Showing a seen line +// twice is the safe failure; hiding an unseen one is not. +func (s *Store) CatchUp(workspaceID, viewerID string, since, until time.Time, scope CatchUpScope) (*CatchUp, error) { + out := &CatchUp{ + Finished: capSection(nil), + NewAndReopened: capSection(nil), + Comments: capSection(nil), + Agents: CatchUpAgents{Groups: []CatchUpAgentGroup{}}, + } + if scope.seesNothing() { + return out, nil + } + sinceStr := since.UTC().Format(time.RFC3339) + untilStr := until.UTC().Format(time.RFC3339) + + colls, err := s.resolveReportCollections(workspaceID, ReportOptions{}) + if err != nil { + return nil, err + } + finished, reopened, err := s.catchUpTransitions(workspaceID, viewerID, sinceStr, untilStr, scope, colls) + if err != nil { + return nil, err + } + out.Finished = capSection(finished) + + created, err := s.catchUpCreated(workspaceID, viewerID, sinceStr, untilStr, scope) + if err != nil { + return nil, err + } + out.NewAndReopened = capSection(mergeNewAndReopened(created, reopened)) + + comments, err := s.catchUpComments(workspaceID, viewerID, sinceStr, untilStr, scope) + if err != nil { + return nil, err + } + out.Comments = capSection(comments) + + agents, err := s.catchUpAgents(workspaceID, sinceStr, untilStr, scope) + if err != nil { + return nil, err + } + out.Agents = agents + return out, nil +} + +// catchUpTransitions returns the in-window finishes (positive terminal) and +// reopens (terminal to non-terminal), one line per item at its latest such +// transition, newest first, with the viewer's own transitions left out. +// +// status_transitions has no actor, so the actor is read from the activity +// row the same write logged: same item, action updated or moved, same +// second (lead ruling, day 13). A transition is the viewer's only when such +// a row exists and every such row is the viewer's. No row, or another +// person's row in that second, keeps the line: showing a user their own +// change is the safe failure, hiding someone else's is not. Debounced update +// rows move their created_at forward, so an update followed by another edit +// by the same writer misses here and is kept. +func (s *Store) catchUpTransitions(workspaceID, viewerID, sinceStr, untilStr string, scope CatchUpScope, colls []reportCollection) (finished, reopened []CatchUpLine, err error) { + finished, reopened = []CatchUpLine{}, []CatchUpLine{} + byColl := make(map[string]reportCollection, len(colls)) + collIDs := make([]string, 0, len(colls)) + for _, c := range colls { + if c.doneKey == "" || len(c.allTerminals) == 0 { + continue + } + byColl[c.id] = c + collIDs = append(collIDs, c.id) + } + if len(collIDs) == 0 { + return finished, reopened, nil + } + scopeSQL, scopeArgs := scope.clause("i") + ph := make([]string, len(collIDs)) + args := []any{workspaceID, sinceStr, untilStr} + for i, id := range collIDs { + ph[i] = "?" + args = append(args, id) + } + args = append(args, scopeArgs...) + // The actor columns come from the activity rows of the same second. A + // LEFT JOIN, so a transition with no row survives with nulls; several + // rows give several result rows, folded below. + rows, err := s.db.Query(s.q(` + SELECT st.item_id, st.collection_id, st.field_key, st.from_status, st.to_status, st.created_at, + c.prefix, i.item_number, i.title, c.slug, + a.user_id, COALESCE(u.name, ''), COALESCE(a.actor, ''), `+s.activityMetaText("a")+` + FROM status_transitions st + JOIN items i ON i.id = st.item_id AND i.deleted_at IS NULL + JOIN collections c ON c.id = i.collection_id AND c.deleted_at IS NULL + LEFT JOIN activities a ON a.document_id = st.item_id AND a.action IN ('updated', 'moved') + AND a.created_at = st.created_at + LEFT JOIN users u ON u.id = a.user_id + WHERE st.workspace_id = ? AND st.created_at >= ? AND st.created_at <= ? + AND st.collection_id IN (`+strings.Join(ph, ",")+`) + AND i.collection_id = st.collection_id`+scopeSQL+` + ORDER BY st.created_at DESC, st.id`), args...) + if err != nil { + return nil, nil, fmt.Errorf("catch-up transitions: %w", err) + } + defer rows.Close() + + type key struct{ item, field, from, to, at string } + type agg struct { + line CatchUpLine + kind string // "finished" | "reopened" + anyRow bool + otherRow bool + otherName string + agent string + } + order := []key{} + seen := map[key]*agg{} + for rows.Next() { + var itemID, collID, field, from, to, at, prefix, title, slug, actorName, actorKind, meta string + var num int + var userID *string + if err := rows.Scan(&itemID, &collID, &field, &from, &to, &at, &prefix, &num, &title, &slug, + &userID, &actorName, &actorKind, &meta); err != nil { + return nil, nil, fmt.Errorf("scan catch-up transition: %w", err) + } + c := byColl[collID] + if field != c.doneKey { + continue + } + kind := "" + switch { + case containsFold(c.positiveTerminals, to): + kind = "finished" + case containsFold(c.allTerminals, from) && !containsFold(c.allTerminals, to): + kind = "reopened" + default: + continue + } + k := key{itemID, field, from, to, at} + g, ok := seen[k] + if !ok { + g = &agg{kind: kind, line: CatchUpLine{ + Ref: fmt.Sprintf("%s-%d", prefix, num), Title: title, Collection: slug, + What: kind, At: at, itemID: itemID, + }} + seen[k] = g + order = append(order, k) + } + if userID == nil && actorKind == "" { + continue // the LEFT JOIN found no activity row + } + g.anyRow = true + if userID == nil || *userID != viewerID { + g.otherRow = true + if g.otherName == "" { + g.otherName = actorName + } + } + if actorKind == "agent" && g.agent == "" { + g.agent = models.AgentNameFromMetadata(meta) + } + } + if err := rows.Err(); err != nil { + return nil, nil, fmt.Errorf("catch-up transitions: %w", err) + } + + // One line per item per kind, at its newest kept transition (rows are + // newest first, so the first kept one wins). + done := map[string]bool{} + for _, k := range order { + g := seen[k] + if g.anyRow && !g.otherRow { + continue // the viewer's own + } + dk := g.kind + "\x00" + k.item + if done[dk] { + continue + } + done[dk] = true + g.line.Actor = g.otherName + g.line.Agent = g.agent + if g.kind == "finished" { + finished = append(finished, g.line) + } else { + reopened = append(reopened, g.line) + } + } + return finished, reopened, nil +} + +// catchUpCreated returns items created in the window by someone other than +// the viewer, newest first. System collections (conventions, playbooks) are +// how the workspace is run, not its work, and are left out. An item with no +// recorded creator is kept. +func (s *Store) catchUpCreated(workspaceID, viewerID, sinceStr, untilStr string, scope CatchUpScope) ([]CatchUpLine, error) { + scopeSQL, scopeArgs := scope.clause("i") + args := append([]any{s.dialect.BoolToInt(false), workspaceID, sinceStr, untilStr, viewerID}, scopeArgs...) + rows, err := s.db.Query(s.q(` + SELECT i.id, c.prefix, i.item_number, i.title, c.slug, i.created_at, COALESCE(u.name, ''), COALESCE(i.created_by, '') + FROM items i + JOIN collections c ON c.id = i.collection_id AND c.deleted_at IS NULL AND c.is_system = ? + LEFT JOIN users u ON u.id = i.created_by_user_id + WHERE i.workspace_id = ? AND i.deleted_at IS NULL + AND i.created_at >= ? AND i.created_at <= ? + AND (i.created_by_user_id IS NULL OR i.created_by_user_id <> ?)`+scopeSQL+` + ORDER BY i.created_at DESC, i.id`), args...) + if err != nil { + return nil, fmt.Errorf("catch-up created: %w", err) + } + defer rows.Close() + out := []CatchUpLine{} + for rows.Next() { + var l CatchUpLine + var num int + var prefix, createdBy string + if err := rows.Scan(&l.itemID, &prefix, &num, &l.Title, &l.Collection, &l.At, &l.Actor, &createdBy); err != nil { + return nil, fmt.Errorf("scan catch-up created: %w", err) + } + l.Ref = fmt.Sprintf("%s-%d", prefix, num) + l.What = "created" + out = append(out, l) + } + return out, rows.Err() +} + +// mergeNewAndReopened interleaves the two lists newest first, one line per +// item: an item created and then reopened in the same window shows once, as +// its newest event. +func mergeNewAndReopened(created, reopened []CatchUpLine) []CatchUpLine { + all := append(append([]CatchUpLine{}, created...), reopened...) + sort.SliceStable(all, func(a, b int) bool { return all[a].At > all[b].At }) + seen := map[string]bool{} + out := []CatchUpLine{} + for _, l := range all { + if seen[l.itemID] { + continue + } + seen[l.itemID] = true + out = append(out, l) + } + return out +} + +// catchUpComments returns comments by others in the window on items the +// viewer created or is assigned to (Dave, v1), one line per item at its +// newest such comment. Deleted comments are left out. +func (s *Store) catchUpComments(workspaceID, viewerID, sinceStr, untilStr string, scope CatchUpScope) ([]CatchUpLine, error) { + scopeSQL, scopeArgs := scope.clause("i") + args := append([]any{workspaceID, sinceStr, untilStr, viewerID, viewerID, viewerID}, scopeArgs...) + rows, err := s.db.Query(s.q(` + SELECT i.id, c.prefix, i.item_number, i.title, c.slug, cm.created_at, COALESCE(u.name, cm.author, '') + FROM comments cm + JOIN items i ON i.id = cm.item_id AND i.deleted_at IS NULL + JOIN collections c ON c.id = i.collection_id AND c.deleted_at IS NULL + LEFT JOIN users u ON u.id = cm.user_id + WHERE cm.workspace_id = ? AND cm.deleted_at IS NULL + AND cm.created_at >= ? AND cm.created_at <= ? + AND (cm.user_id IS NULL OR cm.user_id <> ?) + AND (i.created_by_user_id = ? OR i.assigned_user_id = ?)`+scopeSQL+` + ORDER BY cm.created_at DESC, cm.id`), args...) + if err != nil { + return nil, fmt.Errorf("catch-up comments: %w", err) + } + defer rows.Close() + out := []CatchUpLine{} + seen := map[string]bool{} + for rows.Next() { + var l CatchUpLine + var num int + var prefix string + if err := rows.Scan(&l.itemID, &prefix, &num, &l.Title, &l.Collection, &l.At, &l.Actor); err != nil { + return nil, fmt.Errorf("scan catch-up comment: %w", err) + } + if seen[l.itemID] { + continue + } + seen[l.itemID] = true + l.Ref = fmt.Sprintf("%s-%d", prefix, num) + l.What = "commented" + out = append(out, l) + } + return out, rows.Err() +} + +// catchUpAgents groups the window's agent activity by agent name. Agent rows +// carry the HUMAN's user_id, so the viewer's own agents are deliberately +// kept: their work is exactly what a returning person wants to see. The name +// is read with the renderer's accessor, in Go, so a malformed metadata row +// is one unnamed row rather than a failed query. +func (s *Store) catchUpAgents(workspaceID, sinceStr, untilStr string, scope CatchUpScope) (CatchUpAgents, error) { + out := CatchUpAgents{Groups: []CatchUpAgentGroup{}} + scopeSQL, scopeArgs := scope.clause("i") + args := append([]any{workspaceID, sinceStr, untilStr}, scopeArgs...) + rows, err := s.db.Query(s.q(` + SELECT a.action, a.created_at, `+s.activityMetaText("a")+`, i.id, c.prefix, i.item_number, i.title, c.slug + FROM activities a + JOIN items i ON i.id = a.document_id AND i.deleted_at IS NULL + JOIN collections c ON c.id = i.collection_id AND c.deleted_at IS NULL + WHERE a.workspace_id = ? AND a.actor = 'agent' + AND a.created_at >= ? AND a.created_at <= ?`+scopeSQL+` + ORDER BY a.created_at DESC, a.id`), args...) + if err != nil { + return out, fmt.Errorf("catch-up agents: %w", err) + } + defer rows.Close() + type grp struct { + g CatchUpAgentGroup + items map[string]bool + lines []CatchUpLine + } + byName := map[string]*grp{} + order := []string{} + for rows.Next() { + var action, at, meta, itemID, prefix, title, slug string + var num int + if err := rows.Scan(&action, &at, &meta, &itemID, &prefix, &num, &title, &slug); err != nil { + return out, fmt.Errorf("scan catch-up agent row: %w", err) + } + name := models.AgentNameFromMetadata(meta) + if name == "" { + name = "agent" + } + g, ok := byName[name] + if !ok { + g = &grp{g: CatchUpAgentGroup{Agent: name}, items: map[string]bool{}} + byName[name] = g + order = append(order, name) + } + g.g.Changes++ + if g.items[itemID] { + continue + } + g.items[itemID] = true + g.lines = append(g.lines, CatchUpLine{ + Ref: fmt.Sprintf("%s-%d", prefix, num), Title: title, Collection: slug, + What: action, Agent: name, At: at, itemID: itemID, + }) + } + if err := rows.Err(); err != nil { + return out, fmt.Errorf("catch-up agents: %w", err) + } + // Groups in order of their newest activity (rows are newest first). + for _, name := range order { + g := byName[name] + g.g.Items = len(g.lines) + g.g.Lines = capSection(g.lines).Lines + out.Groups = append(out.Groups, g.g) + } + out.Total = len(out.Groups) + return out, nil +} + +// CatchUpItemLines renders the given items as section lines, limited to +// scope, in the order given. Ids that are deleted, invisible or unknown are +// dropped, so the section's total counts only what the viewer may see. +func (s *Store) CatchUpItemLines(workspaceID string, itemIDs []string, what map[string]string, scope CatchUpScope) (CatchUpSection, error) { + if len(itemIDs) == 0 || scope.seesNothing() { + return capSection(nil), nil + } + scopeSQL, scopeArgs := scope.clause("i") + ph := make([]string, len(itemIDs)) + args := []any{workspaceID} + for i, id := range itemIDs { + ph[i] = "?" + args = append(args, id) + } + args = append(args, scopeArgs...) + rows, err := s.db.Query(s.q(` + SELECT i.id, c.prefix, i.item_number, i.title, c.slug + FROM items i + JOIN collections c ON c.id = i.collection_id AND c.deleted_at IS NULL + WHERE i.workspace_id = ? AND i.deleted_at IS NULL + AND i.id IN (`+strings.Join(ph, ",")+`)`+scopeSQL), args...) + if err != nil { + return CatchUpSection{}, fmt.Errorf("catch-up item lines: %w", err) + } + defer rows.Close() + byID := map[string]CatchUpLine{} + for rows.Next() { + var l CatchUpLine + var num int + var prefix string + if err := rows.Scan(&l.itemID, &prefix, &num, &l.Title, &l.Collection); err != nil { + return CatchUpSection{}, fmt.Errorf("scan catch-up item line: %w", err) + } + l.Ref = fmt.Sprintf("%s-%d", prefix, num) + l.What = what[l.itemID] + byID[l.itemID] = l + } + if err := rows.Err(); err != nil { + return CatchUpSection{}, fmt.Errorf("catch-up item lines: %w", err) + } + lines := []CatchUpLine{} + for _, id := range itemIDs { + if l, ok := byID[id]; ok { + lines = append(lines, l) + } + } + return capSection(lines), nil +} + +// activityMetaText reads activities.metadata as text on both backends: it is +// TEXT on SQLite and JSONB on Postgres, where a COALESCE to the empty string +// would cast that string to JSON and fail the query. +func (s *Store) activityMetaText(alias string) string { + if s.dialect.Driver() == DriverPostgres { + return "COALESCE(" + alias + ".metadata::text, '')" + } + return "COALESCE(" + alias + ".metadata, '')" +} + +func containsFold(list []string, v string) bool { + for _, x := range list { + if strings.EqualFold(x, v) { + return true + } + } + return false +} diff --git a/internal/store/migrations/149_workspace_read_cursors.sql b/internal/store/migrations/149_workspace_read_cursors.sql new file mode 100644 index 000000000..dfa3b4263 --- /dev/null +++ b/internal/store/migrations/149_workspace_read_cursors.sql @@ -0,0 +1,23 @@ +-- Migration 149: the per-user, per-workspace read cursor behind the +-- "Since you were last here" card (TASK-3615, DOC-3616). +-- +-- seen_at is the moment the user last pressed Mark caught up in this +-- workspace. It moves ONLY on that action (Dave, 10-17); reading the card +-- never writes. With no row, the card's window is the 14-day ceiling, and +-- the first Mark caught up creates the row. +-- +-- seen_at is fixed-width UTC ("2006-01-02T15:04:05.000000000Z", the +-- decisionTime format), so TEXT comparison against other timestamps in +-- that format is time comparison. +-- +-- Keyed on users and workspaces, not workspace_members, so a guest can hold +-- one, as with user_workspace_tabs (098). + +CREATE TABLE IF NOT EXISTS workspace_read_cursors ( + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + workspace_id TEXT NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, + seen_at TEXT NOT NULL, + PRIMARY KEY (user_id, workspace_id) +); + +CREATE INDEX IF NOT EXISTS idx_workspace_read_cursors_workspace ON workspace_read_cursors(workspace_id); diff --git a/internal/store/nulcolumns.go b/internal/store/nulcolumns.go index 47046df88..281c7c886 100644 --- a/internal/store/nulcolumns.go +++ b/internal/store/nulcolumns.go @@ -420,6 +420,9 @@ var nulExcluded = map[string]string{ "item_patches.workspace_id": "server-minted workspace UUID from the resolved item (PLAN-3578 step 3)", "item_patches.version_id": "server-minted version UUID from the same transaction (PLAN-3578 step 3)", "item_patches.created_at": "server timestamp from now() (PLAN-3578 step 3)", + "workspace_read_cursors.user_id": "the authenticated user's server-minted UUID (TASK-3615, migration 149)", + "workspace_read_cursors.workspace_id": "server-minted workspace UUID from the resolved request (TASK-3615)", + "workspace_read_cursors.seen_at": "server timestamp from now() in the fixed-width decisionTime format (TASK-3615); never from a request", "idempotency_keys.workspace_id": "server-minted workspace UUID from the resolved request (TASK-3585, migration 139)", "idempotency_keys.user_id": "the authenticated user's server-minted UUID, or empty before setup (TASK-3585)", "idempotency_keys.route": "fixed server constant naming the handler (TASK-3585); never from a request", diff --git a/internal/store/pgmigrations/119_workspace_read_cursors.sql b/internal/store/pgmigrations/119_workspace_read_cursors.sql new file mode 100644 index 000000000..74850cb4a --- /dev/null +++ b/internal/store/pgmigrations/119_workspace_read_cursors.sql @@ -0,0 +1,12 @@ +-- Postgres mirror of SQLite migration 149_workspace_read_cursors.sql +-- (TASK-3615): the per-user, per-workspace read cursor. See that file for +-- the rationale. + +CREATE TABLE IF NOT EXISTS workspace_read_cursors ( + user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + workspace_id TEXT NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE, + seen_at TEXT NOT NULL, + PRIMARY KEY (user_id, workspace_id) +); + +CREATE INDEX IF NOT EXISTS idx_workspace_read_cursors_workspace ON workspace_read_cursors(workspace_id); diff --git a/internal/store/read_cursors.go b/internal/store/read_cursors.go new file mode 100644 index 000000000..00fe41654 --- /dev/null +++ b/internal/store/read_cursors.go @@ -0,0 +1,46 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + "time" +) + +// ReadCursor returns when userID last marked workspaceID caught up +// (TASK-3615). found is false when the user never has: the catch-up window +// then falls back to its ceiling, and reading never creates the row. +func (s *Store) ReadCursor(userID, workspaceID string) (seenAt time.Time, found bool, err error) { + var raw string + err = s.db.QueryRow(s.q( + `SELECT seen_at FROM workspace_read_cursors WHERE user_id = ? AND workspace_id = ?`), + userID, workspaceID).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return time.Time{}, false, nil + } + if err != nil { + return time.Time{}, false, fmt.Errorf("read cursor: %w", err) + } + t, err := time.Parse(decisionTimeFormat, raw) + if err != nil { + return time.Time{}, false, fmt.Errorf("read cursor: parse seen_at %q: %w", raw, err) + } + return t, true, nil +} + +// MarkCaughtUp moves userID's cursor in workspaceID to at, creating it on +// the first call. The cursor never moves backwards: an older at (a request +// that raced a newer one) leaves the stored value alone. Returns the cursor +// as stored. +func (s *Store) MarkCaughtUp(userID, workspaceID string, at time.Time) (time.Time, error) { + v := decisionTime(at) + if _, err := s.db.Exec(s.q( + `INSERT INTO workspace_read_cursors (user_id, workspace_id, seen_at) VALUES (?, ?, ?) + ON CONFLICT (user_id, workspace_id) DO UPDATE SET seen_at = excluded.seen_at + WHERE excluded.seen_at > workspace_read_cursors.seen_at`), + userID, workspaceID, v); err != nil { + return time.Time{}, fmt.Errorf("mark caught up: %w", err) + } + seen, _, err := s.ReadCursor(userID, workspaceID) + return seen, err +}