multica/server/internal/handler/activity.go

package handler

import (
	"context"
	"encoding/json"
	"net/http"
	"sort"

	"github.com/go-chi/chi/v5"
	"github.com/jackc/pgx/v5/pgtype"
	"github.com/multica-ai/multica/server/internal/util"
	db "github.com/multica-ai/multica/server/pkg/db/generated"
)

// TimelineEntry represents a single entry in the issue timeline, which can be
// either an activity log record or a comment.
type TimelineEntry struct {
	Type string `json:"type"` // "activity" or "comment"
	ID   string `json:"id"`

	ActorType string `json:"actor_type"`
	ActorID   string `json:"actor_id"`
	CreatedAt string `json:"created_at"`
	// Display-only identity is hydrated from the global user row for member
	// actors. It remains available after the member leaves this workspace;
	// actor_type + actor_id stay the durable attribution keys.
	ActorName      string `json:"actor_name,omitempty"`
	ActorAvatarURL string `json:"actor_avatar_url,omitempty"`

	// Activity-only fields
	Action  *string         `json:"action,omitempty"`
	Details json.RawMessage `json:"details,omitempty"`

	// Comment-only fields
	Content     *string `json:"content,omitempty"`
	ParentID    *string `json:"parent_id,omitempty"`
	UpdatedAt   *string `json:"updated_at,omitempty"`
	Revision    int64   `json:"revision,omitempty"`
	CommentType *string `json:"comment_type,omitempty"`
	// Set only on comments produced by a quick action run. Unforgeable: there
	// is no request field for it on the generic comment endpoint.
	QuickActionID  *string              `json:"quick_action_id,omitempty"`
	Reactions      []ReactionResponse   `json:"reactions,omitempty"`
	Attachments    []AttachmentResponse `json:"attachments,omitempty"`
	ResolvedAt     *string              `json:"resolved_at,omitempty"`
	ResolvedByType *string              `json:"resolved_by_type,omitempty"`
	ResolvedByID   *string              `json:"resolved_by_id,omitempty"`
	SourceTaskID   *string              `json:"source_task_id,omitempty"`
	// Set only on a tombstone: a comment deleted while it still had replies.
	DeletedAt *string `json:"deleted_at,omitempty"`
	// Supplements lists every running turn this comment steered; the single
	// supplement_* fields mirror the first receipt for older clients.
	Supplements             []CommentSupplementResponse `json:"supplements,omitempty"`
	SupplementTaskID        string                      `json:"supplement_task_id,omitempty"`
	SupplementStatus        string                      `json:"supplement_status,omitempty"`
	SupplementFailureReason *string                     `json:"supplement_failure_reason,omitempty"`
	SupplementDeliveredAt   *string                     `json:"supplement_delivered_at,omitempty"`
}

// timelineHardCap bounds the per-issue timeline payload. Sized as a defensive
// safety net, not a UX page window: see commentHardCap in comment.go for the
// data-shape rationale (#1929). A variable only so the cap tests can shrink it,
// like commentHardCap; nothing outside tests assigns it.
var timelineHardCap = 2000

// timelineProbeLimit reads one row past the cap so "we hit the cap" can be
// distinguished from "the issue happens to have exactly timelineHardCap rows".
// Without the probe row an issue sitting exactly on the boundary would report a
// complete timeline as truncated and pay a needless ancestor-backfill query.
func timelineProbeLimit() int32 { return int32(timelineHardCap) + 1 }

// Truncation is signalled with a response header rather than a body field
// because the unpaginated response is a bare JSON array (TimelineEntriesSchema =
// z.array(TimelineEntrySchema) in packages/core/api/schemas.ts) with nowhere to
// put a flag. The header is additive: existing clients keep validating.
//
// The value names which kinds were truncated ("activity", "comment", or
// "activity,comment") because the two caps are independent — in practice it is
// almost always activity alone.
//
// There is deliberately no companion "window from" header. An earlier revision
// emitted one, but TimestampToString is second-precision RFC3339 while the real
// ordering key is (created_at, id) at full precision, so it could not be used to
// resume a read without skipping or repeating rows inside a shared second. A
// resumable cursor needs to be opaque and carry both halves; that is worth
// designing when there is a consumer, not shipping as a lossy approximation.
//
// Exported so the CORS layer can reference the same identifier
// (corsExposedHeaders in server/cmd/server/router.go). A custom response header
// is invisible to browser JS unless explicitly exposed, so a rename here that
// did not reach the CORS list would silently switch the signal back off.
const HeaderTimelineTruncated = "X-Timeline-Truncated"

// truncatedKinds renders the X-Timeline-Truncated value, "" when nothing was
// truncated.
func truncatedKinds(comments, activities bool) string {
	switch {
	case comments && activities:
		return "activity,comment"
	case comments:
		return "comment"
	case activities:
		return "activity"
	default:
		return ""
	}
}

// takeNewest trims a timelineProbeLimit read down to timelineHardCap and reports
// whether the probe row proved older rows exist. rows must be ascending, so the
// newest timelineHardCap entries are the tail.
func takeNewest[T any](rows []T) ([]T, bool) {
	if len(rows) <= timelineHardCap {
		return rows, false
	}
	return rows[len(rows)-timelineHardCap:], true
}

// timelinePaginatedResponse mirrors the wrapper shape produced by the prior
// cursor-paginated ListTimeline (#2128). It is preserved as a backward-compat
// surface for installed Desktop builds and stale Web bundles between #2128 and
// #1929 that send `?limit=`/`?before=`/`?after=`/`?around=` and parse the
// response with the old TimelinePageSchema (entries + cursors). Cursors are
// always nil and `has_more_after` is always false: the new server returns the
// whole timeline in one shot. `has_more_before` is now truthful — it reports the
// hard-cap clamp — instead of being hardcoded false.
type timelinePaginatedResponse struct {
	Entries       []TimelineEntry `json:"entries"`
	NextCursor    *string         `json:"next_cursor"`
	PrevCursor    *string         `json:"prev_cursor"`
	HasMoreBefore bool            `json:"has_more_before"`
	HasMoreAfter  bool            `json:"has_more_after"`
	TargetIndex   *int            `json:"target_index,omitempty"`
}

// ListTimeline returns the full issue timeline (comments + activities merged).
// Two response shapes coexist for boundary compatibility (#1929):
//
//   - No pagination params → flat ASC `TimelineEntry[]`. Matches the legacy
//     desktop contract (Multica.app ≤ v0.2.25) and the new client.
//   - Any of `limit` / `before` / `after` / `around` present → wrapped object
//     with DESC entries + null cursors + has_more_after=false. Matches what a
//     stale v0.2.26+ build expects when it parses the response with
//     TimelinePageSchema; cursor-walking is now a no-op so the client just
//     sees a single full page.
//
// Both shapes carry the same set of entries — paging and ordering differ.
// Time-based pagination was removed because it split reply threads at page
// boundaries, and at observed data sizes (p99 ~30 comments per issue) the
// cursor machinery was pure overhead.
//
// When the hard cap fires each list is independently reduced to its newest
// entries and X-Timeline-Truncated names which kinds were affected. Comment
// threads cut by the window are completed afterwards within a bounded context
// budget; a thread that cannot be completed is omitted as one unit (MUL-5492).
func (h *Handler) ListTimeline(w http.ResponseWriter, r *http.Request) {
	id := chi.URLParam(r, "id")
	issue, ok := h.loadIssueForUser(w, r, id)
	if !ok {
		return
	}
	ctx := r.Context()

	comments, err := h.Queries.ListCommentsForIssue(ctx, db.ListCommentsForIssueParams{
		IssueID:     issue.ID,
		WorkspaceID: issue.WorkspaceID,
		Limit:       timelineProbeLimit(),
	})
	if err != nil {
		writeError(w, http.StatusInternalServerError, "failed to list comments")
		return
	}
	activities, err := h.Queries.ListActivitiesForIssue(ctx, db.ListActivitiesForIssueParams{
		IssueID: issue.ID,
		Limit:   timelineProbeLimit(),
	})
	if err != nil {
		writeError(w, http.StatusInternalServerError, "failed to list activities")
		return
	}

	q := r.URL.Query()
	wantWrapped := q.Get("limit") != "" || q.Get("before") != "" ||
		q.Get("after") != "" || q.Get("around") != ""

	// Each list is capped independently and reports its own truncation. The two
	// are deliberately NOT clamped to a shared floor.
	//
	// An earlier revision did clamp them, to guarantee the returned window was a
	// contiguous correctly-interleaved slice with no region holding only one of
	// the two kinds. That traded the wrong way round. Comments are human-paced
	// (p99 ~30, max ever observed ~1.1k) so they essentially never reach the cap,
	// while activity is machine-paced and reaches it routinely — so the shared
	// floor was almost always the ACTIVITY floor cutting away comments that had
	// been fetched successfully and would have rendered fine. It deleted real
	// content to buy a cosmetic property, and on a busy issue with thirty
	// comments it was pure loss.
	//
	// Not clamping costs only activity density in the older part of the range,
	// which is metadata, not content — and it is reported rather than hidden.
	// It also keeps the comment set from being cut at an arbitrary row, which is
	// why completeCommentThreads repairs any affected thread below.
	comments, commentsTruncated := takeNewest(comments)
	activities, activitiesTruncated := takeNewest(activities)

	// Timeline clients derive resolution bars, author lists, and folded counts
	// from the comments in this response. A parent-only repair would therefore
	// make a partial thread look complete. Restore all missing siblings and
	// descendants for affected roots within the shared context budget; if the
	// walk cannot prove a thread complete, omit that thread as one unit.
	if commentsTruncated {
		var err error
		comments, err = h.completeCommentThreads(ctx, issue.ID, issue.WorkspaceID, comments)
		if err != nil {
			writeError(w, http.StatusInternalServerError, "failed to complete comment threads")
			return
		}
	}

	if kinds := truncatedKinds(commentsTruncated, activitiesTruncated); kinds != "" {
		w.Header().Set(HeaderTimelineTruncated, kinds)
	}

	entries := h.mergeTimeline(r, comments, activities, !wantWrapped)
	// The current-member directory deliberately excludes departed members, but
	// timeline attribution must remain readable after they leave. Hydrate only
	// the member ids already present in this authorised issue response; lookup
	// failure is display-only and must not make the timeline unavailable.
	h.hydrateTimelineMemberActors(ctx, entries)
	if entries == nil {
		entries = []TimelineEntry{}
	}

	if wantWrapped {
		resp := timelinePaginatedResponse{
			Entries:       entries,
			HasMoreBefore: commentsTruncated || activitiesTruncated,
		}
		// `around=<id>`: locate the anchor in the DESC slice so the legacy
		// client can scroll-to-highlight without a follow-up request.
		if anchor := q.Get("around"); anchor != "" {
			for i, e := range entries {
				if e.ID == anchor {
					idx := i
					resp.TargetIndex = &idx
					break
				}
			}
		}
		writeJSON(w, http.StatusOK, resp)
		return
	}

	writeJSON(w, http.StatusOK, entries)
}

// mergeTimeline merges comments and activities and returns them sorted by
// (created_at, id). When ascending=true, oldest first (the new flat-array
// contract); otherwise newest first (the wrapped legacy contract).
func (h *Handler) mergeTimeline(r *http.Request, comments []db.Comment, activities []db.ActivityLog, ascending bool) []TimelineEntry {
	out := make([]TimelineEntry, 0, len(comments)+len(activities))
	out = append(out, h.commentsToEntries(r, comments)...)
	for _, a := range activities {
		out = append(out, activityToEntry(a))
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].CreatedAt != out[j].CreatedAt {
			if ascending {
				return out[i].CreatedAt < out[j].CreatedAt
			}
			return out[i].CreatedAt > out[j].CreatedAt
		}
		if ascending {
			return out[i].ID < out[j].ID
		}
		return out[i].ID > out[j].ID
	})
	return out
}

// commentsToEntries fetches reactions + attachments for the given comments in
// one batch each and returns enriched TimelineEntry slices preserving order.
func (h *Handler) commentsToEntries(r *http.Request, comments []db.Comment) []TimelineEntry {
	if len(comments) == 0 {
		return nil
	}
	ids := make([]pgtype.UUID, len(comments))
	for i, c := range comments {
		ids[i] = c.ID
	}
	reactions := h.groupReactions(r, ids)
	attachments := h.groupAttachments(r, ids)
	supplements := h.listCommentSupplements(r.Context(), comments[0].WorkspaceID, ids)

	out := make([]TimelineEntry, len(comments))
	for i, c := range comments {
		content := c.Content
		commentType := c.Type
		updatedAt := timestampToString(c.UpdatedAt)
		cid := uuidToString(c.ID)
		out[i] = TimelineEntry{
			Type:           "comment",
			ID:             cid,
			ActorType:      c.AuthorType,
			ActorID:        uuidToString(c.AuthorID),
			Content:        &content,
			CommentType:    &commentType,
			QuickActionID:  uuidToPtr(c.QuickActionID),
			ParentID:       uuidToPtr(c.ParentID),
			CreatedAt:      timestampToString(c.CreatedAt),
			UpdatedAt:      &updatedAt,
			Revision:       c.Revision,
			Reactions:      reactions[cid],
			Attachments:    attachments[cid],
			ResolvedAt:     timestampToPtr(c.ResolvedAt),
			ResolvedByType: textToPtr(c.ResolvedByType),
			ResolvedByID:   uuidToPtr(c.ResolvedByID),
			SourceTaskID:   uuidToPtr(c.SourceTaskID),
			DeletedAt:      timestampToPtr(c.DeletedAt),
		}
		if receipts := supplements[cid]; len(receipts) > 0 {
			out[i].Supplements = receipts
			out[i].SupplementTaskID = receipts[0].TaskID
			out[i].SupplementStatus = receipts[0].Status
			out[i].SupplementFailureReason = receipts[0].FailureReason
			out[i].SupplementDeliveredAt = receipts[0].DeliveredAt
		}
	}
	return out
}

func activityToEntry(a db.ActivityLog) TimelineEntry {
	action := a.Action
	actorType := ""
	if a.ActorType.Valid {
		actorType = a.ActorType.String
	}
	return TimelineEntry{
		Type:      "activity",
		ID:        uuidToString(a.ID),
		ActorType: actorType,
		ActorID:   uuidToString(a.ActorID),
		Action:    &action,
		Details:   a.Details,
		CreatedAt: timestampToString(a.CreatedAt),
	}
}

// hydrateTimelineMemberActors adds display identity for member-authored rows
// without changing the active-member directory's semantics. The ids came from
// comments/activity rows on an issue loadIssueForUser already authorised, so
// this creates no arbitrary user lookup surface. The query is bounded by the
// timeline hard caps and runs once for the whole response, never once per row.
func (h *Handler) hydrateTimelineMemberActors(ctx context.Context, entries []TimelineEntry) {
	seen := make(map[string]struct{})
	ids := make([]pgtype.UUID, 0)
	for i := range entries {
		entry := &entries[i]
		if entry.ActorType != "member" || entry.ActorID == "" {
			continue
		}
		if _, ok := seen[entry.ActorID]; ok {
			continue
		}
		id, err := util.ParseUUID(entry.ActorID)
		if err != nil {
			continue
		}
		seen[entry.ActorID] = struct{}{}
		ids = append(ids, id)
	}
	if len(ids) == 0 {
		return
	}

	users, err := h.Queries.GetUsersByIDs(ctx, ids)
	if err != nil {
		return
	}
	byID := make(map[string]db.GetUsersByIDsRow, len(users))
	for _, user := range users {
		byID[uuidToString(user.ID)] = user
	}
	for i := range entries {
		entry := &entries[i]
		if entry.ActorType != "member" {
			continue
		}
		user, ok := byID[entry.ActorID]
		if !ok {
			continue
		}
		entry.ActorName = user.Name
		if user.AvatarUrl.Valid {
			entry.ActorAvatarURL = h.resolveAvatarURL(user.AvatarUrl.String)
		}
	}
}

// AssigneeFrequencyEntry represents how often a user assigns to a specific target.
type AssigneeFrequencyEntry struct {
	AssigneeType string `json:"assignee_type"`
	AssigneeID   string `json:"assignee_id"`
	Frequency    int64  `json:"frequency"`
}

// GetAssigneeFrequency returns assignee usage frequency for the current user,
// combining data from assignee change activities and initial issue assignments.
func (h *Handler) GetAssigneeFrequency(w http.ResponseWriter, r *http.Request) {
	userID, ok := requireUserID(w, r)
	if !ok {
		return
	}
	workspaceID := h.resolveWorkspaceID(r)

	// Aggregate frequency from both data sources.
	freq := map[string]int64{} // key: "type:id"

	// Source 1: assignee_changed activities by this user.
	activityCounts, err := h.Queries.CountAssigneeChangesByActor(r.Context(), db.CountAssigneeChangesByActorParams{
		WorkspaceID: parseUUID(workspaceID),
		ActorID:     parseUUID(userID),
	})
	if err != nil {
		writeError(w, http.StatusInternalServerError, "failed to get assignee frequency")
		return
	}
	for _, row := range activityCounts {
		aType, _ := row.AssigneeType.(string)
		aID, _ := row.AssigneeID.(string)
		if aType != "" && aID != "" {
			freq[aType+":"+aID] += row.Frequency
		}
	}

	// Source 2: issues created by this user with an assignee.
	issueCounts, err := h.Queries.CountCreatedIssueAssignees(r.Context(), db.CountCreatedIssueAssigneesParams{
		WorkspaceID: parseUUID(workspaceID),
		CreatorID:   parseUUID(userID),
	})
	if err != nil {
		writeError(w, http.StatusInternalServerError, "failed to get assignee frequency")
		return
	}
	for _, row := range issueCounts {
		if !row.AssigneeType.Valid || !row.AssigneeID.Valid {
			continue
		}
		key := row.AssigneeType.String + ":" + uuidToString(row.AssigneeID)
		freq[key] += row.Frequency
	}

	// Build sorted response.
	result := make([]AssigneeFrequencyEntry, 0, len(freq))
	for key, count := range freq {
		// Split "type:id" — type is always "member" or "agent" (no colons).
		var aType, aID string
		for i := 0; i < len(key); i++ {
			if key[i] == ':' {
				aType = key[:i]
				aID = key[i+1:]
				break
			}
		}
		result = append(result, AssigneeFrequencyEntry{
			AssigneeType: aType,
			AssigneeID:   aID,
			Frequency:    count,
		})
	}
	sort.Slice(result, func(i, j int) bool {
		return result[i].Frequency > result[j].Frequency
	})

	writeJSON(w, http.StatusOK, result)
}