import {
  FieldPath,
  Timestamp,
  type Query,
  type QueryDocumentSnapshot,
} from "@google-cloud/firestore";
import type Redis from "ioredis";
import { logError, logInfo } from "@/lib/logger";
import { getEventTrackingFirestore } from "@/lib/firestore/event-tracking";
import {
  COLLECTION_LIST_TTL_MS,
  getEventCollections,
  getFieldMap,
  getMaxAggregationDocs,
  getTimestampKind,
} from "./event-analytics.config";
import { mapEventDoc, type EventRecord } from "./event-analytics.mapper";
import type { EventMemberProfile } from "@/internal/repository/event-analytics/member-lookup.repo";

/**
 * Read-only analytics over the booking events in the `event-tracking` Firestore
 * database.
 *
 * The database has one top-level collection per event type — `add_to_cart`,
 * `booking_hold_initiated`, `purchase` — rather than one collection with a type
 * field. Every read therefore fans out across the selected collections and
 * merges the results here; Firestore cannot union differently-named top-level
 * collections itself (a collection-group query only unifies collections that
 * share an ID).
 *
 * IMPORTANT: every method here is a read. This service must never write to
 * Firestore — the events are owned by the producing system.
 */

export type EventFilters = {
  from?: string;
  to?: string;
  /** Which collections to read. Empty/absent means all of them. */
  eventTypes?: string[];
  memberId?: string;
  currency?: string;
  property?: string;
  source?: string;
  search?: string;
};

/** Revenue is never summed across currencies — each stands on its own. */
export type RevenueByCurrency = { currency: string; revenue: number; events: number };

export type EventSummary = {
  totalEvents: number;
  uniqueMembers: number;
  uniqueEventTypes: number;
  revenueByCurrency: RevenueByCurrency[];
  /**
   * Loyalty points earned across the filtered set, and how many events carried
   * any. Most events in this dataset have no `amount`/`currency`, so points are
   * usually the only quantity present.
   */
  totalPoints: number;
  pointsEvents: number;
  topEventType: { eventType: string; events: number } | null;
  firstEventAt: string | null;
  lastEventAt: string | null;
  /** True when the scan hit the aggregation ceiling — figures are a partial view. */
  truncated: boolean;
  scannedDocs: number;
};

export type MemberAnalyticsRow = {
  memberId: string;
  /**
   * Name and email as written on the event, which the current producer never
   * sets. The real values come from the member DB and are attached as `profile`
   * by the controller — these stay so a producer that starts sending them is
   * picked up without a change here.
   */
  memberName: string | null;
  memberEmail: string | null;
  /** Filled in from the Updot member DB. Null when the number has no match. */
  profile?: EventMemberProfile | null;
  events: number;
  eventTypes: number;
  revenueByCurrency: RevenueByCurrency[];
  points: number;
  /** Distinct properties this member generated events at. */
  properties: string[];
  /**
   * Per event type, how many events this member produced. Keyed by collection
   * name, so a funnel reads directly off one row:
   * add_to_cart 4 -> booking_hold_initiated 2 -> purchase 1.
   */
  eventCounts: Record<string, number>;
  firstSeen: string | null;
  lastSeen: string | null;
};

export type EventTypeAnalyticsRow = {
  eventType: string;
  events: number;
  uniqueMembers: number;
  revenueByCurrency: RevenueByCurrency[];
  points: number;
  /** Share of all events in the filtered set, 0-100, one decimal. */
  sharePct: number;
};

export type EventAnalyticsAggregate = {
  summary: EventSummary;
  members: MemberAnalyticsRow[];
  eventTypes: EventTypeAnalyticsRow[];
  /** Every collection in the database, for the filter dropdown. */
  availableEventTypes: string[];
  availableCurrencies: string[];
  availableProperties: string[];
  availableSources: string[];
  /** Activity per day across every member, for the module timeline. */
  byDay: DayActivity[];
  /** Per-property roll-up across every member. */
  byProperty: PropertyActivity[];
  /** Per-source roll-up. `sourceOfEvent` cuts across event types. */
  bySource: SourceActivity[];
};

export type DayActivity = {
  day: string;
  events: number;
  points: number;
  members: number;
};

export type PropertyActivity = {
  property: string;
  events: number;
  points: number;
  members: number;
};

export type SourceActivity = {
  source: string;
  events: number;
  points: number;
  members: number;
};

/** One member's complete activity, for the member detail page. */
export type MemberBreakdown = {
  records: EventRecord[];
  /** True when the member has more events than the page will read. */
  truncated: boolean;
  byDay: {
    day: string;
    events: number;
    points: number;
    /** Events per type on that day, keyed by collection name. */
    byType: Record<string, number>;
  }[];
  byProperty: { property: string; events: number; points: number }[];
  bySource: { source: string; events: number; points: number }[];
};

/**
 * Ceiling on one member's events read for the detail page.
 *
 * Generous — a single member producing more than this is not a reporting case —
 * but present, because "one member cannot have many events" is an assumption
 * about data we do not control.
 */
const MEMBER_RECORD_LIMIT = 2000;

const CACHE_TTL_SECONDS = 300;

/**
 * Bump whenever `EventAnalyticsAggregate` gains or changes a field.
 *
 * The cached value is a serialised aggregate, so an entry written by an earlier
 * build is still valid JSON and still deserialises — it is simply missing the
 * new fields. Adding `byDay` without bumping this served aggregates with no
 * `byDay`, and the dashboard reported "0 days with activity" against data that
 * was there. The version makes an older entry unreachable rather than wrong.
 */
const AGGREGATE_SCHEMA_VERSION = 3;
const CACHE_PREFIX = `event-analytics:agg:v${AGGREGATE_SCHEMA_VERSION}:`;

/**
 * Does a cached payload have everything this build expects?
 *
 * Belt and braces behind the version: the version only helps if whoever adds a
 * field remembers to bump it, and the failure mode when they forget is silent
 * wrong numbers rather than an error. This catches it on the next read instead.
 */
function isCurrentAggregate(value: unknown): value is EventAnalyticsAggregate {
  if (typeof value !== "object" || value === null) return false;
  const v = value as Partial<EventAnalyticsAggregate>;
  return (
    Array.isArray(v.members) &&
    Array.isArray(v.eventTypes) &&
    Array.isArray(v.byDay) &&
    Array.isArray(v.byProperty) &&
    Array.isArray(v.bySource) &&
    typeof v.summary === "object" &&
    v.summary !== null
  );
}

export class EventAnalyticsService {
  private static instance: EventAnalyticsService | null = null;

  static getInstance(): EventAnalyticsService {
    if (!EventAnalyticsService.instance) {
      EventAnalyticsService.instance = new EventAnalyticsService();
    }
    return EventAnalyticsService.instance;
  }

  private collectionCache: { names: string[]; at: number } | null = null;

  /**
   * The event types available, i.e. the database's top-level collections.
   *
   * Discovered at runtime unless pinned by EVENT_TRACKING_COLLECTIONS, so a new
   * event type the producer starts writing shows up without a deploy. Cached
   * briefly — `listCollections()` is a round trip and the set changes rarely.
   */
  async listEventTypes(): Promise<string[]> {
    const configured = getEventCollections();
    if (configured) return configured;

    const now = Date.now();
    if (this.collectionCache && now - this.collectionCache.at < COLLECTION_LIST_TTL_MS) {
      return this.collectionCache.names;
    }

    const refs = await getEventTrackingFirestore().listCollections();
    const names = refs.map((r) => r.id).sort();
    this.collectionCache = { names, at: now };
    return names;
  }

  /** The collections a filter set actually needs read. */
  private async resolveCollections(filters: EventFilters): Promise<string[]> {
    const all = await this.listEventTypes();
    const wanted = filters.eventTypes?.filter(Boolean) ?? [];
    if (!wanted.length) return all;
    // Intersect rather than trust the input: an unknown collection name would
    // otherwise produce a query against a collection that does not exist,
    // which Firestore answers with an empty page rather than an error.
    const known = new Set(all);
    return wanted.filter((name) => known.has(name));
  }

  // ── query construction ────────────────────────────────────────────────────

  /**
   * Convert an ISO bound into whatever type the timestamp field is stored as.
   * See `getTimestampKind` — a type mismatch here silently matches nothing.
   *
   * The `string` case is compared lexicographically by Firestore, so the bound
   * must be padded to the producer's precision rather than round-tripped
   * through `toISOString()`. The producer writes six fractional digits and no
   * timezone ("2026-08-07T17:09:05.822772"); a three-digit `...000Z` lower
   * bound would sort *after* an event at exactly midnight ('Z' > '0') and
   * silently drop it.
   */
  private bound(iso: string, edge: "start" | "end"): Timestamp | string | number {
    const date = new Date(iso);
    switch (getTimestampKind()) {
      case "string":
        return (
          date.toISOString().replace(/\.\d+Z$/, "") +
          (edge === "start" ? ".000000" : ".999999")
        );
      case "number":
        return date.getTime();
      default:
        return Timestamp.fromDate(date);
    }
  }

  /**
   * Normalise a user-supplied bound. A date-only `to` is widened to the end of
   * that day, so "to: 2026-08-14" includes everything that happened on the 14th
   * rather than only the midnight instant.
   */
  private parseBound(value: string, edge: "start" | "end"): string | null {
    const dateOnly = /^\d{4}-\d{2}-\d{2}$/.test(value);
    const iso = dateOnly
      ? `${value}T${edge === "start" ? "00:00:00.000" : "23:59:59.999"}Z`
      : value;
    const d = new Date(iso);
    return Number.isNaN(d.getTime()) ? null : d.toISOString();
  }

  /**
   * Build the query for one collection.
   *
   * Everything Firestore can evaluate is pushed down. Two exceptions:
   * `eventTypes`, which is expressed by choosing collections rather than by a
   * clause, and `search`, which Firestore cannot do at all (no substring
   * matching) and so is applied in memory by the caller.
   */
  private buildQuery(collection: string, filters: EventFilters): Query {
    const map = getFieldMap();
    let query: Query = getEventTrackingFirestore().collection(collection);

    if (filters.from) {
      const from = this.parseBound(filters.from, "start");
      if (from) query = query.where(map.occurredAt, ">=", this.bound(from, "start"));
    }
    if (filters.to) {
      const to = this.parseBound(filters.to, "end");
      if (to) query = query.where(map.occurredAt, "<=", this.bound(to, "end"));
    }
    if (filters.memberId) {
      query = query.where(map.memberId, "==", filters.memberId);
    }
    if (filters.currency) {
      query = query.where(map.currency, "==", filters.currency.toUpperCase());
    }
    if (filters.property) {
      query = query.where(map.propertyName, "==", filters.property);
    }
    if (filters.source) {
      query = query.where(map.source, "==", filters.source);
    }

    return query;
  }

  /** In-memory predicate for the parts Firestore cannot express. */
  private matchesSearch(record: EventRecord, search?: string): boolean {
    if (!search) return true;
    const needle = search.trim().toLowerCase();
    if (!needle) return true;
    return [
      record.memberName,
      record.memberEmail,
      record.memberId,
      record.bookingRef,
      record.eventType,
      record.source,
      record.propertyName,
      record.experienceName,
      record.memberOffer,
    ].some((v) => v?.toLowerCase().includes(needle));
  }

  // ── aggregation ───────────────────────────────────────────────────────────

  private cacheKey(filters: EventFilters): string {
    // Sorted keys so two equivalent filter objects share one cache entry.
    const normalised = {
      from: filters.from ?? "",
      to: filters.to ?? "",
      eventTypes: [...(filters.eventTypes ?? [])].sort(),
      memberId: filters.memberId ?? "",
      currency: filters.currency ?? "",
      property: filters.property ?? "",
      source: filters.source ?? "",
      search: filters.search ?? "",
    };
    return CACHE_PREFIX + Buffer.from(JSON.stringify(normalised)).toString("base64url");
  }

  /**
   * Every aggregate the dashboard needs, from a single pass over the matching
   * documents in each selected collection.
   *
   * One pass rather than four: per-member, per-event-type and per-currency
   * revenue all need the same documents, and Firestore has no GROUP BY, so
   * splitting them would multiply the read cost by four for identical data. The
   * result is cached, so paging or switching tabs does not re-scan.
   *
   * `select()` narrows the fields transferred. It does not reduce the number of
   * documents billed, but it materially cuts payload and parse time.
   */
  async computeAggregates(
    filters: EventFilters,
    redis?: Redis,
  ): Promise<EventAnalyticsAggregate> {
    const key = this.cacheKey(filters);

    if (redis) {
      try {
        const cached = await redis.get(key);
        if (cached) {
          const parsed = JSON.parse(cached) as unknown;
          if (isCurrentAggregate(parsed)) return parsed;
          // Shape from an older build: drop it and recompute rather than serve
          // a payload with fields this build will read as empty.
          await redis.del(key).catch(() => undefined);
        }
      } catch (err) {
        // A cache miss must never be fatal; fall through to a live read.
        logError(err, "event-analytics: redis read failed");
      }
    }

    const map = getFieldMap();
    const maxDocs = getMaxAggregationDocs();
    const allEventTypes = await this.listEventTypes();
    const collections = await this.resolveCollections(filters);

    type MemberAcc = {
      memberId: string;
      memberName: string | null;
      memberEmail: string | null;
      events: number;
      types: Set<string>;
      revenue: Map<string, { revenue: number; events: number }>;
      points: number;
      properties: Set<string>;
      eventCounts: Record<string, number>;
      firstSeen: string | null;
      lastSeen: string | null;
    };
    type TypeAcc = {
      eventType: string;
      events: number;
      members: Set<string>;
      revenue: Map<string, { revenue: number; events: number }>;
      points: number;
    };

    const members = new Map<string, MemberAcc>();
    const types = new Map<string, TypeAcc>();
    const currencyTotals = new Map<string, { revenue: number; events: number }>();
    const memberIds = new Set<string>();
    const properties = new Set<string>();
    const sources = new Set<string>();
    /*
     * Day and property roll-ups, folded in the same pass as everything else.
     *
     * Member sets rather than counters: a member active on the same day twice is
     * one member, and a count would report two. The sets are dropped for their
     * size once the fold finishes.
     */
    const days = new Map<string, { events: number; points: number; members: Set<string> }>();
    const propertyStats = new Map<
      string,
      { events: number; points: number; members: Set<string> }
    >();
    const sourceStats = new Map<
      string,
      { events: number; points: number; members: Set<string> }
    >();

    let scanned = 0;
    let matched = 0;
    let totalPoints = 0;
    let pointsEvents = 0;
    let firstEventAt: string | null = null;
    let lastEventAt: string | null = null;
    /*
     * Timestamps that could not be read, reported once per pass.
     *
     * A null `occurredAt` is invisible in the UI — the row just shows "—" — but
     * it silently breaks date filtering, sorting and first/last seen for that
     * event. Since the field name is the producer's, not ours, the log names the
     * keys the document actually had so a rename is diagnosable from one request
     * instead of a debugging session.
     */
    let missingTimestamps = 0;
    let missingTimestampSample: string[] | null = null;

    const addRevenue = (
      target: Map<string, { revenue: number; events: number }>,
      currency: string | null,
      revenue: number | null,
    ) => {
      if (revenue === null || currency === null) return;
      const bucket = target.get(currency) ?? { revenue: 0, events: 0 };
      bucket.revenue += revenue;
      bucket.events += 1;
      target.set(currency, bucket);
    };

    /*
     * Collections are scanned one after another, not concurrently.
     *
     * The ceiling is a budget across all of them, and a sequential scan can stop
     * the moment it is spent. Running them in parallel would read up to
     * `maxDocs` from every collection before anything could be counted.
     */
    for (const collection of collections) {
      if (scanned > maxDocs) break;

      /*
       * Deliberately not a projection query.
       *
       * This used to `select()` the handful of fields the aggregation reads.
       * That bought almost nothing — these documents are nine scalar fields with
       * no payload blob, so the transfer saving is noise — while adding a way
       * for a field to come back missing and silently null out `occurredAt`,
       * which breaks first/last seen and date filtering without any error.
       * Reading whole documents is both simpler and one less thing to be wrong.
       */
      const query = this.buildQuery(collection, filters).limit(
        maxDocs + 1 - scanned,
      );

      const stream = query.stream() as AsyncIterable<QueryDocumentSnapshot>;

      for await (const doc of stream) {
        scanned += 1;
        if (scanned > maxDocs) break;

        const record = mapEventDoc(doc, collection);
        if (!this.matchesSearch(record, filters.search)) continue;
        matched += 1;

        if (record.occurredAt) {
          if (!firstEventAt || record.occurredAt < firstEventAt) firstEventAt = record.occurredAt;
          if (!lastEventAt || record.occurredAt > lastEventAt) lastEventAt = record.occurredAt;
        } else {
          missingTimestamps += 1;
          missingTimestampSample ??= Object.keys(doc.data() ?? {});
        }

        if (record.occurredAt) {
          const day = record.occurredAt.slice(0, 10);
          const acc = days.get(day) ?? { events: 0, points: 0, members: new Set<string>() };
          acc.events += 1;
          acc.points += record.points ?? 0;
          if (record.memberId) acc.members.add(record.memberId);
          days.set(day, acc);
        }

        if (record.propertyName) {
          properties.add(record.propertyName);
          const acc = propertyStats.get(record.propertyName) ?? {
            events: 0,
            points: 0,
            members: new Set<string>(),
          };
          acc.events += 1;
          acc.points += record.points ?? 0;
          if (record.memberId) acc.members.add(record.memberId);
          propertyStats.set(record.propertyName, acc);
        }
        if (record.source) {
          sources.add(record.source);
          const acc = sourceStats.get(record.source) ?? {
            events: 0,
            points: 0,
            members: new Set<string>(),
          };
          acc.events += 1;
          acc.points += record.points ?? 0;
          if (record.memberId) acc.members.add(record.memberId);
          sourceStats.set(record.source, acc);
        }
        if (record.points !== null) {
          totalPoints += record.points;
          pointsEvents += 1;
        }

        // Per event type (= per collection).
        const typeAcc: TypeAcc = types.get(collection) ?? {
          eventType: collection,
          events: 0,
          members: new Set<string>(),
          revenue: new Map(),
          points: 0,
        };
        typeAcc.events += 1;
        if (record.memberId) typeAcc.members.add(record.memberId);
        typeAcc.points += record.points ?? 0;
        addRevenue(typeAcc.revenue, record.currency, record.revenue);
        types.set(collection, typeAcc);

        // Per member. Events with no member id still count towards totals and
        // event-type rows, but cannot appear in the member breakdown.
        if (record.memberId) {
          memberIds.add(record.memberId);
          const acc: MemberAcc = members.get(record.memberId) ?? {
            memberId: record.memberId,
            memberName: null,
            memberEmail: null,
            events: 0,
            types: new Set<string>(),
            revenue: new Map(),
            points: 0,
            properties: new Set<string>(),
            eventCounts: {},
            firstSeen: null,
            lastSeen: null,
          };
          acc.events += 1;
          acc.types.add(collection);
          acc.eventCounts[collection] = (acc.eventCounts[collection] ?? 0) + 1;
          acc.memberName ??= record.memberName;
          acc.memberEmail ??= record.memberEmail;
          acc.points += record.points ?? 0;
          if (record.propertyName) acc.properties.add(record.propertyName);
          addRevenue(acc.revenue, record.currency, record.revenue);
          if (record.occurredAt) {
            if (!acc.firstSeen || record.occurredAt < acc.firstSeen) acc.firstSeen = record.occurredAt;
            if (!acc.lastSeen || record.occurredAt > acc.lastSeen) acc.lastSeen = record.occurredAt;
          }
          members.set(record.memberId, acc);
        }

        addRevenue(currencyTotals, record.currency, record.revenue);
      }
    }

    if (missingTimestamps) {
      logInfo(
        `event-analytics: ${missingTimestamps} of ${matched} events had no readable "${map.occurredAt}"; ` +
          `date filters and first/last seen are wrong for those. Fields actually present: ` +
          `${(missingTimestampSample ?? []).join(", ") || "(none)"}. ` +
          `Fix with EVENT_TRACKING_FIELD_MAP={"occurredAt":"<real field>"}.`,
      );
    }

    const truncated = scanned > maxDocs;
    if (truncated) {
      logInfo(
        `event-analytics: aggregation hit the ${maxDocs}-document ceiling; results reported as truncated`,
      );
    }

    const toRevenueList = (
      m: Map<string, { revenue: number; events: number }>,
    ): RevenueByCurrency[] =>
      [...m.entries()]
        .map(([currency, v]) => ({ currency, revenue: v.revenue, events: v.events }))
        .sort((a, b) => b.revenue - a.revenue);

    const eventTypeRows: EventTypeAnalyticsRow[] = [...types.values()]
      .map((t) => ({
        eventType: t.eventType,
        events: t.events,
        uniqueMembers: t.members.size,
        revenueByCurrency: toRevenueList(t.revenue),
        points: t.points,
        sharePct: matched ? Number(((t.events / matched) * 100).toFixed(1)) : 0,
      }))
      .sort((a, b) => b.events - a.events);

    const memberRows: MemberAnalyticsRow[] = [...members.values()]
      .map((m) => ({
        memberId: m.memberId,
        memberName: m.memberName,
        memberEmail: m.memberEmail,
        events: m.events,
        eventTypes: m.types.size,
        revenueByCurrency: toRevenueList(m.revenue),
        points: m.points,
        properties: [...m.properties].sort(),
        eventCounts: m.eventCounts,
        firstSeen: m.firstSeen,
        lastSeen: m.lastSeen,
      }))
      .sort((a, b) => b.events - a.events);

    const result: EventAnalyticsAggregate = {
      summary: {
        totalEvents: matched,
        uniqueMembers: memberIds.size,
        uniqueEventTypes: types.size,
        revenueByCurrency: toRevenueList(currencyTotals),
        totalPoints,
        pointsEvents,
        topEventType: eventTypeRows.length
          ? { eventType: eventTypeRows[0].eventType, events: eventTypeRows[0].events }
          : null,
        firstEventAt,
        lastEventAt,
        truncated,
        scannedDocs: Math.min(scanned, maxDocs),
      },
      members: memberRows,
      eventTypes: eventTypeRows,
      // Every collection, not just the ones with results — the dropdown must
      // still offer a type the current filter happens to exclude.
      availableEventTypes: allEventTypes,
      availableCurrencies: [...currencyTotals.keys()].sort(),
      availableProperties: [...properties].sort(),
      availableSources: [...sources].sort(),
      // Ascending: a timeline reads left to right.
      byDay: [...days.entries()]
        .sort((a, b) => a[0].localeCompare(b[0]))
        .map(([day, v]) => ({
          day,
          events: v.events,
          points: v.points,
          members: v.members.size,
        })),
      byProperty: [...propertyStats.entries()]
        .map(([property, v]) => ({
          property,
          events: v.events,
          points: v.points,
          members: v.members.size,
        }))
        .sort((a, b) => b.events - a.events),
      bySource: [...sourceStats.entries()]
        .map(([source, v]) => ({
          source,
          events: v.events,
          points: v.points,
          members: v.members.size,
        }))
        .sort((a, b) => b.events - a.events),
    };

    if (redis) {
      try {
        await redis.set(key, JSON.stringify(result), "EX", CACHE_TTL_SECONDS);
      } catch (err) {
        logError(err, "event-analytics: redis write failed");
      }
    }

    return result;
  }

  // ── detail records ────────────────────────────────────────────────────────

  /**
   * One page of detail records, newest first, merged across collections.
   *
   * A k-way merge: each collection contributes its next `pageSize` documents,
   * the merged head is taken, and each collection's cursor advances only past
   * what was actually consumed. That keeps a single global ordering without
   * reading a collection further than the page needs.
   *
   * Paged by cursor rather than `offset`: Firestore bills every skipped document
   * on an offset query, so deep paging on a large collection gets steadily more
   * expensive and slower. A cursor page costs the same at page 500 as at page 1.
   *
   * A `search` filter is applied after the fetch, so a page may come back
   * shorter than `pageSize`; `hasMore` and the cursor still drive paging
   * correctly.
   */
  async listEvents(params: {
    filters: EventFilters;
    pageSize: number;
    cursor?: string;
  }): Promise<{ records: EventRecord[]; nextCursor: string | null; hasMore: boolean }> {
    const map = getFieldMap();
    const collections = await this.resolveCollections(params.filters);
    const cursors = decodeCursor(params.cursor);

    // Each collection is read independently, so these run concurrently — unlike
    // the aggregation pass, there is no shared budget to spend down in order.
    const pages = await Promise.all(
      collections.map(async (collection) => {
        let query = this.buildQuery(collection, params.filters)
          .orderBy(map.occurredAt, "desc")
          .orderBy(FieldPath.documentId(), "desc");

        const cursor = cursors?.[collection];
        if (cursor) query = query.startAfter(cursor.value, cursor.id);

        const snapshot = await query.limit(params.pageSize).get();
        return snapshot.docs.map((doc) => ({
          collection,
          doc,
          record: mapEventDoc(doc, collection),
        }));
      }),
    );

    // Merge on the same key the per-collection queries sorted by, so the merged
    // order is the true global order.
    const merged = pages
      .flat()
      .sort((a, b) => {
        const at = a.record.occurredAt ?? "";
        const bt = b.record.occurredAt ?? "";
        if (at !== bt) return at < bt ? 1 : -1;
        return a.doc.id < b.doc.id ? 1 : -1;
      });

    const taken = merged.slice(0, params.pageSize);

    /*
     * A collection is exhausted only if it returned fewer rows than asked for.
     * If every collection came back short and the merge fit in one page, there
     * is nothing after this one.
     */
    const anyFull = pages.some((p) => p.length === params.pageSize);
    const hasMore = merged.length > params.pageSize || anyFull;

    // Advance each collection's cursor to the last of ITS documents that made
    // this page. A collection that contributed nothing keeps its old cursor, so
    // its unread documents are not skipped.
    const nextCursors: Record<string, RawCursor> = { ...(cursors ?? {}) };
    for (const item of taken) {
      nextCursors[item.collection] = {
        value: item.doc.get(map.occurredAt),
        id: item.doc.id,
      };
    }

    const records = taken
      .map((item) => item.record)
      .filter((r) => this.matchesSearch(r, params.filters.search));

    return {
      records,
      nextCursor: hasMore ? encodeCursor(nextCursors) : null,
      hasMore,
    };
  }

  /**
   * Everything one member did, in full.
   *
   * A whole-history read rather than a paged one: with `memberId` pushed down
   * into every collection query, one member's events are a few hundred documents
   * at most, and having them all in hand is what lets the detail page show a
   * timeline, a per-property split and the complete record list without three
   * more round trips.
   *
   * Bounded anyway — an unbounded read is an unbounded read regardless of how
   * unlikely the worst case is — and the bound is reported, never silent.
   */
  async getMemberBreakdown(
    filters: EventFilters,
    limit = MEMBER_RECORD_LIMIT,
  ): Promise<MemberBreakdown> {
    const collections = await this.resolveCollections(filters);

    const perCollection = await Promise.all(
      collections.map(async (collection) => {
        const snapshot = await this.buildQuery(collection, filters)
          .limit(limit + 1)
          .get();
        return snapshot.docs.map((doc) => mapEventDoc(doc, collection));
      }),
    );

    const all = perCollection
      .flat()
      .filter((r) => this.matchesSearch(r, filters.search))
      // Newest first, matching the record list everywhere else. Undated events
      // sort last rather than being dropped — they still happened.
      .sort((a, b) => (b.occurredAt ?? "").localeCompare(a.occurredAt ?? ""));

    const truncated = all.length > limit;
    const records = truncated ? all.slice(0, limit) : all;

    type DayAcc = { events: number; points: number; byType: Record<string, number> };
    const days = new Map<string, DayAcc>();
    const properties = new Map<string, { events: number; points: number }>();
    const sources = new Map<string, { events: number; points: number }>();

    for (const record of records) {
      if (record.occurredAt) {
        const day = record.occurredAt.slice(0, 10);
        const acc = days.get(day) ?? { events: 0, points: 0, byType: {} };
        acc.events += 1;
        acc.points += record.points ?? 0;
        acc.byType[record.eventType] = (acc.byType[record.eventType] ?? 0) + 1;
        days.set(day, acc);
      }

      if (record.propertyName) {
        const acc = properties.get(record.propertyName) ?? { events: 0, points: 0 };
        acc.events += 1;
        acc.points += record.points ?? 0;
        properties.set(record.propertyName, acc);
      }

      if (record.source) {
        const acc = sources.get(record.source) ?? { events: 0, points: 0 };
        acc.events += 1;
        acc.points += record.points ?? 0;
        sources.set(record.source, acc);
      }
    }

    return {
      records,
      truncated,
      // Ascending: a timeline reads left to right.
      byDay: [...days.entries()]
        .sort((a, b) => a[0].localeCompare(b[0]))
        .map(([day, v]) => ({ day, ...v })),
      byProperty: [...properties.entries()]
        .map(([property, v]) => ({ property, ...v }))
        .sort((a, b) => b.events - a.events),
      bySource: [...sources.entries()]
        .map(([source, v]) => ({ source, ...v }))
        .sort((a, b) => b.events - a.events),
    };
  }

  /**
   * Total matching documents across the selected collections.
   *
   * Uses Firestore's aggregation `count()`, which is billed in index-entry units
   * rather than document reads and does not stream documents.
   *
   * Not available with a `search` filter — that predicate lives in memory, so
   * the true count requires reading the documents. Callers get `null` and should
   * present the count as unknown rather than show the unfiltered number, which
   * would be wrong.
   */
  async countEvents(filters: EventFilters): Promise<number | null> {
    if (filters.search?.trim()) return null;
    const collections = await this.resolveCollections(filters);
    const counts = await Promise.all(
      collections.map(async (collection) => {
        const snapshot = await this.buildQuery(collection, filters).count().get();
        return snapshot.data().count;
      }),
    );
    return counts.reduce((sum, n) => sum + n, 0);
  }
}

// ── cursor encoding ─────────────────────────────────────────────────────────

type RawCursor = { value: unknown; id: string };
type DecodedCursors = Record<string, { value: Timestamp | string | number; id: string }>;

/**
 * Encode one sort position per collection as a single opaque string.
 *
 * The stored type is carried alongside each value: a cursor built as a Timestamp
 * will not resume a field stored as an ISO string, so the type is preserved from
 * the document that produced it rather than guessed on the way back in.
 */
function encodeCursor(cursors: Record<string, RawCursor>): string {
  const payload: Record<string, { k: string; v: string | number; id: string }> = {};

  for (const [collection, cursor] of Object.entries(cursors)) {
    const raw = cursor.value;
    if (raw instanceof Timestamp) {
      payload[collection] = { k: "ts", v: raw.toDate().toISOString(), id: cursor.id };
    } else if (raw instanceof Date) {
      payload[collection] = { k: "ts", v: raw.toISOString(), id: cursor.id };
    } else if (typeof raw === "number") {
      payload[collection] = { k: "n", v: raw, id: cursor.id };
    } else {
      payload[collection] = { k: "s", v: String(raw ?? ""), id: cursor.id };
    }
  }

  return Buffer.from(JSON.stringify(payload)).toString("base64url");
}

function decodeCursor(cursor?: string): DecodedCursors | null {
  if (!cursor) return null;
  try {
    const payload = JSON.parse(Buffer.from(cursor, "base64url").toString()) as Record<
      string,
      { k: string; v: string | number; id: string }
    >;

    const out: DecodedCursors = {};
    for (const [collection, entry] of Object.entries(payload)) {
      if (entry.k === "ts") {
        out[collection] = { value: Timestamp.fromDate(new Date(entry.v)), id: entry.id };
      } else if (entry.k === "n") {
        out[collection] = { value: Number(entry.v), id: entry.id };
      } else {
        out[collection] = { value: String(entry.v), id: entry.id };
      }
    }
    return out;
  } catch {
    // A corrupt cursor restarts from the first page, which is preferable to
    // failing the request.
    return null;
  }
}