member-console/internal/server/operator_activity.go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo

package server

import (
	"context"
	"database/sql"
	"log/slog"
	"math"
	"sort"

	"git.coopcloud.tech/wiki-cafe/member-console/internal/billing"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/entitlements"
)

// loadRecentActivity assembles the landing-surface unified timeline and
// returns the page p asks for with the number of events the timeline holds.
// Page p of size s lists events (p-1)·s+1 to p·s, and each of them is among
// its own source's first p·s, so loadRecentActivity reads each event source
// (grants, transitions, invoices, payments) to that depth in the timeline's
// order (activityLess), merges the rows, cuts the page and names only the
// page's rows; every event lists on exactly one page. The sources leave out
// the rows a supersession makes redundant themselves (core.superseded_ends,
// core.extension_handoffs), so the events the timeline holds are the sum of
// the sources' counts over their history.
//
// Failures from any individual source are logged and the source is
// dropped — the timeline degrades gracefully rather than 500ing the
// landing surface. An empty result is the legitimate empty state. A page past
// the end is page 1: p.Page is set to 1, which the landing's pager reads, and
// the rows already read hold page 1.
func (h *OperatorHandler) loadRecentActivity(ctx context.Context, p *ListParams) ([]ActivityEvent, int) {
	src := h.readActivitySources(ctx, feedDepth(p))
	rows := feedRows(src)
	sort.SliceStable(rows, func(i, j int) bool { return activityLess(rows[i].ActivityEvent, rows[j].ActivityEvent) })
	total := src.total()
	if int64(p.Offset()) >= total && p.Page > 1 {
		p.Page = 1
	}
	page := feedPage(rows, p)
	return nameFeedRows(page, h.pageNames(ctx, page)), int(total)
}

// feedDepth is how many rows of each source page p needs: the rows of every
// page up to and including it, at most what a query's LIMIT takes.
func feedDepth(p *ListParams) int32 {
	return int32(min(int64(p.Offset())+int64(p.Limit()), math.MaxInt32))
}

// feedPage cuts page p out of the merged rows, which hold every row of the
// timeline down to the end of the page.
func feedPage(rows []feedRow, p *ListParams) []feedRow {
	start := min(int(p.Offset()), len(rows))
	end := min(start+int(p.Limit()), len(rows))
	return rows[start:end]
}

// activitySources is what one read of the four sources returned; a source
// whose read failed is empty.
type activitySources struct {
	grants      []entitlements.ListRecentGrantsRow
	transitions []entitlements.ListRecentTransitionsRow
	invoices    []billing.ListRecentInvoicesRow
	payments    []billing.ListRecentPaymentsRow
}

// readActivitySources reads each source's first depth rows in the
// timeline's order. A failed read is logged and leaves that source empty.
func (h *OperatorHandler) readActivitySources(ctx context.Context, depth int32) activitySources {
	var src activitySources
	var err error
	if src.grants, err = h.EntitlementsQ.ListRecentGrants(ctx, depth); err != nil {
		h.Logger.Warn("activity feed: list recent grants failed", slog.Any("error", err))
	}
	if src.transitions, err = h.EntitlementsQ.ListRecentTransitions(ctx, depth); err != nil {
		h.Logger.Warn("activity feed: list recent transitions failed", slog.Any("error", err))
	}
	if src.invoices, err = h.BillingQ.ListRecentInvoices(ctx, depth); err != nil {
		h.Logger.Warn("activity feed: list recent invoices failed", slog.Any("error", err))
	}
	if src.payments, err = h.BillingQ.ListRecentPayments(ctx, depth); err != nil {
		h.Logger.Warn("activity feed: list recent payments failed", slog.Any("error", err))
	}
	return src
}

// total is the number of events the timeline holds: each source's count over
// its history, which every row it returned carries. A source whose read
// failed counts nothing.
func (src activitySources) total() int64 {
	return sourceCount(src.grants, func(r entitlements.ListRecentGrantsRow) int64 { return r.TotalCount }) +
		sourceCount(src.transitions, func(r entitlements.ListRecentTransitionsRow) int64 { return r.TotalCount }) +
		sourceCount(src.invoices, func(r billing.ListRecentInvoicesRow) int64 { return r.TotalCount }) +
		sourceCount(src.payments, func(r billing.ListRecentPaymentsRow) int64 { return r.TotalCount })
}

// sourceCount is the count over its history that one source's rows carry, 0
// when the source returned none.
func sourceCount[R any](rows []R, count func(R) int64) int64 {
	if len(rows) == 0 {
		return 0
	}
	return count(rows[0])
}

// feedRow is one feed event before it is named: everything the order reads,
// and the grant or transition its summary is written from.
type feedRow struct {
	ActivityEvent
	grant      *entitlements.ListRecentGrantsRow
	transition *entitlements.ListRecentTransitionsRow
}

// feedRows builds one row per event the feed lists, in source order:
// grants, transitions, invoices, payments.
func feedRows(src activitySources) []feedRow {
	rows := make([]feedRow, 0, len(src.grants)+len(src.transitions)+len(src.invoices)+len(src.payments))
	for i := range src.grants {
		rows = append(rows, grantRow(&src.grants[i]))
	}
	for i := range src.transitions {
		rows = append(rows, feedTransitionRow(&src.transitions[i]))
	}
	for _, inv := range src.invoices {
		rows = append(rows, invoiceRow(inv))
	}
	for _, p := range src.payments {
		rows = append(rows, paymentRow(p))
	}
	return rows
}

// grantRow builds the event of one grant, before its summary is written.
func grantRow(g *entitlements.ListRecentGrantsRow) feedRow {
	orgID := ""
	if g.GrantedToOrgID.Valid {
		orgID = g.GrantedToOrgID.UUID.String()
	}
	personID := ""
	if g.GrantedByPersonID.Valid {
		personID = g.GrantedByPersonID.UUID.String()
	}
	return feedRow{grant: g, ActivityEvent: ActivityEvent{
		EventType: "grant_issued",
		Seq:       g.GrantID,
		When:      NewDateTime(g.CreatedAt),
		SortKey:   g.CreatedAt.UnixNano(),
		OrgID:     orgID,
		PersonID:  personID,
		Href:      "/operator/organizations/" + orgID,
	}}
}

// feedTransitionRow builds the event of one tier change, before its summary
// is written.
func feedTransitionRow(t *entitlements.ListRecentTransitionsRow) feedRow {
	personID := ""
	if t.ActorID.Valid {
		personID = t.ActorID.UUID.String()
	}
	return feedRow{transition: t, ActivityEvent: ActivityEvent{
		EventType: "transition",
		Seq:       t.TransitionID,
		When:      NewDateTime(t.EffectiveAt),
		SortKey:   t.EffectiveAt.UnixNano(),
		OrgID:     t.OrgID,
		PersonID:  personID,
		Href:      "/operator/organizations/" + t.OrgID,
	}}
}

// invoiceRow builds the event of one invoice. Invoices and payments are placed
// when they happened, as Stripe reports it (stripe-invoice-reconcile D17),
// never when the console wrote the row; both queries return only rows that
// have that time. Their summaries name nothing, so they are written here.
func invoiceRow(inv billing.ListRecentInvoicesRow) feedRow {
	return feedRow{ActivityEvent: ActivityEvent{
		EventType: "invoice_created",
		Seq:       inv.InvoiceID,
		When:      NewDateTime(inv.IssuedAt.Time),
		SortKey:   inv.IssuedAt.Time.UnixNano(),
		OrgID:     inv.OrgID,
		Summary:   "Invoice " + formatCurrency(inv.AmountDue, inv.Currency) + " · " + inv.Status,
		Href:      "/operator/billing/invoices",
	}}
}

// paymentRow builds the event of one payment, placed as invoiceRow places an
// invoice.
func paymentRow(p billing.ListRecentPaymentsRow) feedRow {
	at := paymentOutcomeTime(p.PaidAt, p.FailedAt).Time
	return feedRow{ActivityEvent: ActivityEvent{
		EventType: "payment_received",
		Seq:       p.PaymentID,
		When:      NewDateTime(at),
		SortKey:   at.UnixNano(),
		OrgID:     p.OrgID,
		Summary:   "Payment " + formatCurrency(p.Amount, p.Currency) + " · " + p.Status,
		Href:      "/operator/billing/payments",
	}}
}

// activityLess orders the feed newest first. Ties (events stamped in the same
// instant, as a seed or a signup burst produces) break on kind, then the
// order the rows were written, the later first, then summary, organization,
// actor and link, so the feed's order is a function of its data, never of
// query arrival order: a pinned demo snapshot stamps whole batches into one
// instant, and synchronized sequential scans vary the arrival order run to
// run. Every event's Seq is its own row's id, so no two events get past it,
// and their summaries, written after the order, never decide. Each source's
// query cuts its rows in this same order (newest first, then the later-written
// row), so a page's read depth never splits a tie the merged order would list
// differently.
func activityLess(a, b ActivityEvent) bool {
	if a.SortKey != b.SortKey {
		return a.SortKey > b.SortKey
	}
	if a.EventType != b.EventType {
		return a.EventType < b.EventType
	}
	if a.Seq != b.Seq {
		return a.Seq > b.Seq
	}
	if a.Summary != b.Summary {
		return a.Summary < b.Summary
	}
	if a.OrgID != b.OrgID {
		return a.OrgID < b.OrgID
	}
	if a.PersonID != b.PersonID {
		return a.PersonID < b.PersonID
	}
	return a.Href < b.Href
}

// nameFeedRows writes each row's summary from its grant or transition, and
// its organization and person names.
func nameFeedRows(rows []feedRow, n feedNames) []ActivityEvent {
	events := make([]ActivityEvent, len(rows))
	for i, r := range rows {
		events[i] = r.named(n)
	}
	return events
}

// named returns the row's event with its summary, organization name and
// person name written.
func (r feedRow) named(n feedNames) ActivityEvent {
	e := r.ActivityEvent
	if g := r.grant; g != nil {
		e.Summary = grantSummary(g, n.product(g.ProductID))
	}
	if t := r.transition; t != nil {
		// ACC-26 / operator-panel-navigation "Tier events read as the
		// composite reads them": the summary uses the SAME vocabulary the
		// organization composite's Tier changes list uses, never the raw
		// transition_type or a bare integer rank.
		fromLabel := n.tier(t.PlanLadderID, t.FromRank)
		toLabel := n.tier(t.PlanLadderID, t.ToRank)
		e.Summary = humanizeTransitionSummary(t.TransitionType, t.ChangeKind, fromLabel, toLabel)
	}
	e.OrgName = n.org(e.OrgID)
	e.PersonName = n.person(e.PersonID)
	return e
}

// grantSummary words a grant's event from its row and its product's name.
func grantSummary(g *entitlements.ListRecentGrantsRow, productName string) string {
	summary := "Issued " + productName + " (" + g.GrantReason + ")"
	if g.ExtendsGrantID.Valid {
		summary = "Extended: " + productName
	}
	// A default the floor restored after a plan ended names that plan, in
	// the words the grants ledger uses (grant-resumption D17); every other
	// grant keeps its summary.
	if len(g.RestoredAfterProductNames) > 0 {
		summary += " after " + joinWithAnd(g.RestoredAfterProductNames) + " ended"
	}
	return summary
}

// pageNames names the page's organizations, people, products and ladders on
// the request's resolver, so the first lookup of each kind reads the whole
// page's ids in one statement.
func (h *OperatorHandler) pageNames(ctx context.Context, page []feedRow) feedNames {
	n := h.names(ctx)
	for _, r := range page {
		n.WantOrgs(r.OrgID)
		n.WantPersons(r.PersonID)
		if r.grant != nil {
			n.WantProducts(r.grant.ProductID)
		}
		if r.transition != nil {
			n.WantTiers(r.transition.PlanLadderID)
		}
	}
	return feedNames{n}
}

// feedNames answers a name the batch did not return as a failed per-row read
// did: an empty name, or a tier by its rank.
type feedNames struct{ n *operatorNames }

// product names a product.
func (f feedNames) product(id string) string {
	name, _ := f.n.Product(id)
	return name
}

// tier names a rank on a ladder as the ladder is now.
func (f feedNames) tier(ladderID string, rank sql.NullInt32) string {
	return tierLabel(f.n, ladderID, rank)
}

// org names an organization.
func (f feedNames) org(id string) string {
	name, _ := f.n.Org(id)
	return name
}

// person names a person.
func (f feedNames) person(id string) string {
	name, _ := f.n.Person(id)
	return name
}