/**
 * Home activity feed writer.
 *
 * Owns `<workspace>/data/home-feed.json`, the daemon-side source of
 * truth for the macOS Home page activity feed.
 *
 * **v2 merge semantics** — the schema collapse to a single
 * `notification` type also collapses the writer's merge rules to a
 * single rule:
 *
 *   - **Same `id` replaces in place**: if an incoming item shares its
 *     `id` with an existing item, replace that item while preserving
 *     its array position so the UI does not jitter on updates.
 *     Otherwise, append. The pre-v2 type-specific branches (digest
 *     replacement by source, thread same-id update, action
 *     append-without-replace, hybrid-author resolution, per-source
 *     action cap) are gone — they were holdovers from a multi-type
 *     vocabulary that no longer exists.
 *
 *   - **TTL filter on read**: `readHomeFeed` drops any item whose
 *     `expiresAt` is in the past. This is a stateless sweep — the
 *     writer does not rewrite the file on read, so concurrent reads
 *     never race the writer. Callers that want auto-expiry must set
 *     `expiresAt` explicitly; the writer does NOT fill in a default.
 *
 * Concurrent writers are coalesced with the exact same "latest wins"
 * pattern as `relationship-state-writer.ts`: at most one compute+write
 * runs at a time, and overlapping calls during an in-flight write all
 * resolve off a single tail write that reflects the final state.
 *
 * Each successful write publishes a `home_feed_updated` SSE event via
 * the in-process `assistantEventHub`, carrying the post-filter count
 * of items with `status === "new"` so subscribers can update unread
 * badges without a full refetch.
 */

import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
import { join } from "node:path";

import { broadcastMessage } from "../runtime/assistant-event-hub.js";
import { getLogger } from "../util/logger.js";
import { getDataDir } from "../util/platform.js";
import {
  type FeedItem,
  type FeedItemStatus,
  type FeedItemUrgency,
  type HomeFeedFile,
  parseFeedFile,
} from "./feed-types.js";

const log = getLogger("home-feed-writer");

/** Filename for the on-disk home feed. Lives under the workspace data dir. */
export const HOME_FEED_FILENAME = "home-feed.json";

/** On-disk file-format version. Bump + migrate if the shape changes. */
export const HOME_FEED_VERSION = 2;

/**
 * Canonical path to the home-feed snapshot
 * (`<workspace>/data/home-feed.json`).
 */
export function getHomeFeedPath(): string {
  return join(getDataDir(), HOME_FEED_FILENAME);
}

/**
 * Read the on-disk feed file, applying the stateless TTL filter.
 *
 * Returns an empty `HomeFeedFile` when the file is missing, unreadable,
 * or has an invalid envelope. Callers never see a throw from this path.
 * Individual items that fail validation are dropped and warn-logged
 * while the rest of the feed survives.
 * Items whose `expiresAt` is in the past are dropped from the returned
 * `items` array but are NOT rewritten to disk; the next append cycle
 * will persist the post-filter view naturally.
 */
export function readHomeFeed(): HomeFeedFile {
  const path = getHomeFeedPath();
  const empty: HomeFeedFile = {
    version: HOME_FEED_VERSION,
    items: [],
    updatedAt: new Date(0).toISOString(),
  };

  if (!existsSync(path)) {
    return empty;
  }

  let raw: unknown;
  try {
    raw = JSON.parse(readFileSync(path, "utf-8"));
  } catch (err) {
    log.warn({ err, path }, "Failed to read home-feed.json; returning empty");
    return empty;
  }

  let parsed: ReturnType<typeof parseFeedFile>;
  try {
    parsed = parseFeedFile(raw);
  } catch (err) {
    log.warn(
      { err, path },
      "home-feed.json failed schema validation; returning empty",
    );
    return empty;
  }

  if (parsed.droppedCount > 0) {
    log.warn(
      { path, droppedCount: parsed.droppedCount },
      "Dropped invalid items from home-feed.json",
    );
  }

  const now = Date.now();
  const items = parsed.items.filter((item) => !isExpired(item, now));
  return {
    version: parsed.version,
    items,
    updatedAt: parsed.updatedAt,
  };
}

/**
 * Append (or merge) a single feed item and persist the result.
 *
 * See the module comment for the precise merge semantics. Never
 * throws — all failures degrade to a warn-log so fire-and-forget
 * callers in the daemon don't need a try/catch wrapper. Concurrent
 * calls are coalesced via the in-module `writeInFlight` / `writeDirty`
 * pattern so at most one write is in flight at a time.
 */
export async function appendFeedItem(item: FeedItem): Promise<void> {
  pendingAppends.push(item);
  return scheduleWrite();
}

/**
 * Update the `status` field of a single feed item by id.
 *
 * Returns the updated `FeedItem` on success, or `null` if no item with
 * the given id exists. This is the path the HTTP route uses when the
 * client marks an item as `"seen"` or `"acted_on"`. Concurrent patches
 * go through the same coalescing queue as `appendFeedItem` so two
 * overlapping status flips can't race each other.
 *
 * The patch is applied inside `runWrite()` so the existence check
 * reads from the same state snapshot the mutation will land on —
 * callers never observe a "phantom success" where we return an
 * updated item for an id that no longer exists on disk by the time
 * the queued write runs.
 */
export async function patchFeedItemStatus(
  id: string,
  status: FeedItemStatus,
): Promise<FeedItem | null> {
  let resolveResult!: (value: FeedItem | null) => void;
  const resultPromise = new Promise<FeedItem | null>((resolve) => {
    resolveResult = resolve;
  });
  pendingPatches.push({ id, status, resolve: resolveResult });
  void scheduleWrite();
  return resultPromise;
}

/**
 * Patch the user-editable copy fields on a single feed item by id.
 *
 * Returns the updated `FeedItem`, or `null` if no item with the given
 * id exists. Used by the `assistant notifications edit` flow. Patches
 * are applied inside the same coalescing queue as appends and status
 * patches so overlapping writes don't race.
 *
 * Only fields explicitly present on `patch` are touched. Pass an empty
 * object and the call is a no-op that returns the existing item (or
 * `null` if the id isn't on disk). A `title` that trims to empty is
 * ignored, so no edit path can strip a title off an existing item.
 */
export interface FeedItemContentPatch {
  title?: string;
  summary?: string;
  urgency?: FeedItemUrgency;
  status?: FeedItemStatus;
}

export async function patchFeedItemContent(
  id: string,
  patch: FeedItemContentPatch,
): Promise<FeedItem | null> {
  let resolveResult!: (value: FeedItem | null) => void;
  const resultPromise = new Promise<FeedItem | null>((resolve) => {
    resolveResult = resolve;
  });
  pendingContentPatches.push({ id, patch, resolve: resolveResult });
  void scheduleWrite();
  return resultPromise;
}

/**
 * Remove the `conversationId` field from all feed items that reference
 * the given conversation. Returns the number of items modified.
 *
 * This is the daemon-side cleanup path invoked when a conversation is
 * deleted — it prevents "Go to Thread" buttons from linking to a
 * conversation that no longer exists. Goes through the same coalescing
 * queue as appends and patches so there are no file-I/O races.
 */
export async function stripConversationIds(
  conversationId: string,
): Promise<number> {
  let resolveResult!: (count: number) => void;
  const resultPromise = new Promise<number>((resolve) => {
    resolveResult = resolve;
  });
  pendingStrips.push({ conversationId, resolve: resolveResult });
  void scheduleWrite();
  return resultPromise;
}

/**
 * Remove the `conversationId` field from ALL feed items that have one.
 * Returns the number of items modified.
 *
 * Used by the clear-all-conversations flow to bulk-invalidate every
 * "Go to Thread" link at once. Uses the same coalescing queue with a
 * `"*"` sentinel that the strip loop in `runWrite` treats as "match
 * all items with a conversationId".
 */
export async function clearAllConversationIds(): Promise<number> {
  return stripConversationIds("*");
}

/**
 * Bulk-flip the status of every feed item currently at one of the
 * `from` statuses to the single `to` status. Returns the count of
 * items whose status was changed. Items already at the target status
 * (or outside `from`) are left untouched, so the returned count is
 * the real "how many did we mutate" figure the caller can surface.
 *
 * Goes through the same coalescing write queue as appends and single
 * patches, so overlapping callers cannot race on the on-disk file.
 * Returns `-1` when the underlying write fails so callers can
 * distinguish a legitimate zero-count from a persistence failure.
 */
export async function bulkSetFeedItemStatus(
  from: readonly FeedItemStatus[],
  to: FeedItemStatus,
  ids?: readonly string[],
): Promise<number> {
  let resolveResult!: (count: number) => void;
  const resultPromise = new Promise<number>((resolve) => {
    resolveResult = resolve;
  });
  pendingBulkStatus.push({ from, to, ids, resolve: resolveResult });
  void scheduleWrite();
  return resultPromise;
}

// ─── Internal: coalescing queue ────────────────────────────────────────

/**
 * Pending operations that land in the next coalesced write cycle.
 * Appends and patches drain together so overlapping callers share a
 * single compute+write tail.
 */
const pendingAppends: FeedItem[] = [];
const pendingPatches: Array<{
  id: string;
  status: FeedItemStatus;
  resolve: (value: FeedItem | null) => void;
}> = [];
const pendingContentPatches: Array<{
  id: string;
  patch: FeedItemContentPatch;
  resolve: (value: FeedItem | null) => void;
}> = [];
const pendingStrips: Array<{
  conversationId: string;
  resolve: (count: number) => void;
}> = [];
const pendingBulkStatus: Array<{
  from: readonly FeedItemStatus[];
  to: FeedItemStatus;
  ids?: readonly string[];
  resolve: (count: number) => void;
}> = [];

let writeInFlight: Promise<void> | null = null;
let writeDirty = false;

/**
 * Enqueue a write cycle. Mirrors the `relationship-state-writer.ts`
 * coalescing pattern exactly: the first caller kicks off a run; any
 * callers that arrive during an in-flight run mark dirty and resolve
 * off the same tail promise, so N overlapping callers produce at most
 * two runs (the initial + one coalesced tail).
 */
function scheduleWrite(): Promise<void> {
  if (writeInFlight) {
    writeDirty = true;
    return writeInFlight;
  }
  writeInFlight = (async () => {
    try {
      await runWrite();
      while (writeDirty) {
        writeDirty = false;
        await runWrite();
      }
    } finally {
      writeInFlight = null;
    }
  })();
  return writeInFlight;
}

/**
 * Drain the pending-operations queue into a fresh on-disk snapshot
 * and publish the SSE event. Never throws — the write error is caught
 * + logged so the coalescing loop can still move on to the next cycle.
 */
async function runWrite(): Promise<void> {
  const appendsToApply = pendingAppends.splice(0, pendingAppends.length);
  const patchesToApply = pendingPatches.splice(0, pendingPatches.length);
  const contentPatchesToApply = pendingContentPatches.splice(
    0,
    pendingContentPatches.length,
  );
  const stripsToApply = pendingStrips.splice(0, pendingStrips.length);
  const bulkStatusToApply = pendingBulkStatus.splice(
    0,
    pendingBulkStatus.length,
  );

  const current = readHomeFeed();
  let items = current.items.slice();

  for (const incoming of appendsToApply) {
    items = mergeIncoming(items, incoming);
  }

  // Track the per-patch result so callers can distinguish an update
  // from an unknown-id no-op. We collect resolvers first and fire them
  // after the write lands so the resolved `FeedItem` matches on-disk
  // state exactly.
  const patchResults: Array<{
    resolve: (v: FeedItem | null) => void;
    value: FeedItem | null;
  }> = [];
  for (const patch of patchesToApply) {
    const idx = items.findIndex((i) => i.id === patch.id);
    if (idx === -1) {
      patchResults.push({ resolve: patch.resolve, value: null });
      continue;
    }
    const updated: FeedItem = { ...items[idx]!, status: patch.status };
    items[idx] = updated;
    patchResults.push({ resolve: patch.resolve, value: updated });
  }

  const contentPatchResults: Array<{
    resolve: (v: FeedItem | null) => void;
    value: FeedItem | null;
  }> = [];
  for (const { id, patch, resolve } of contentPatchesToApply) {
    const idx = items.findIndex((i) => i.id === id);
    if (idx === -1) {
      contentPatchResults.push({ resolve, value: null });
      continue;
    }
    const existing = items[idx]!;
    const updated: FeedItem = { ...existing };
    if (patch.title !== undefined) {
      const trimmed = patch.title.trim();
      // Blank titles are ignored: an item keeps its title once it has one.
      if (trimmed.length > 0) {
        updated.title = trimmed;
      }
    }
    if (patch.summary !== undefined) {
      updated.summary = patch.summary;
    }
    if (patch.urgency !== undefined) {
      updated.urgency = patch.urgency;
    }
    if (patch.status !== undefined) {
      updated.status = patch.status;
    }
    items[idx] = updated;
    contentPatchResults.push({ resolve, value: updated });
  }

  // Strip conversationId from matching items.
  const stripResults: Array<{
    resolve: (count: number) => void;
    count: number;
  }> = [];
  for (const strip of stripsToApply) {
    let count = 0;
    for (let i = 0; i < items.length; i++) {
      const matchAll = strip.conversationId === "*";
      const matchOne = items[i]!.conversationId === strip.conversationId;
      if (matchAll ? items[i]!.conversationId != null : matchOne) {
        items[i] = { ...items[i]!, conversationId: undefined };
        count++;
      }
    }
    stripResults.push({ resolve: strip.resolve, count });
  }

  // Bulk-flip status. Applied after single patches so overlapping
  // single-item patches land first and the bulk pass observes the
  // post-patch statuses.
  const bulkStatusResults: Array<{
    resolve: (count: number) => void;
    count: number;
  }> = [];
  for (const op of bulkStatusToApply) {
    const fromSet = new Set(op.from);
    const idSet = op.ids ? new Set(op.ids) : null;
    let count = 0;
    for (let i = 0; i < items.length; i++) {
      const current = items[i]!;
      if (current.status === op.to) {
        continue;
      }
      if (!fromSet.has(current.status)) {
        continue;
      }
      if (idSet && !idSet.has(current.id)) {
        continue;
      }
      items[i] = { ...current, status: op.to };
      count++;
    }
    bulkStatusResults.push({ resolve: op.resolve, count });
  }

  items.sort(compareFeedItems);

  const updatedAt = new Date().toISOString();
  const next: HomeFeedFile = {
    version: HOME_FEED_VERSION,
    items,
    updatedAt,
  };

  let wrote = false;
  try {
    const path = getHomeFeedPath();
    mkdirSync(getDataDir(), { recursive: true });
    writeFileSync(path, JSON.stringify(next, null, 2), "utf-8");
    wrote = true;
    log.info({ path, items: items.length }, "Wrote home-feed.json");
  } catch (err) {
    log.warn({ err }, "Failed to write home-feed.json");
  }

  if (wrote) {
    const newItemCount = items.filter((i) => i.status === "new").length;
    publishHomeFeedUpdated(updatedAt, newItemCount);
  }

  // Resolve pending patch and strip promises AFTER we've emitted the
  // SSE event so callers awaiting `patchFeedItemStatus` or
  // `stripConversationIds` observe a fully consistent world: the
  // on-disk file, the SSE event, and the returned value all reflect
  // the same write.
  //
  // If the write failed, resolve patch promises with `null` and strip
  // promises with `0` — the state was not persisted, and callers must
  // not report success when the underlying write failed.
  for (const { resolve, value } of patchResults) {
    resolve(wrote ? value : null);
  }
  for (const { resolve, value } of contentPatchResults) {
    resolve(wrote ? value : null);
  }
  for (const { resolve, count } of stripResults) {
    resolve(wrote ? count : 0);
  }
  for (const { resolve, count } of bulkStatusResults) {
    resolve(wrote ? count : -1);
  }
}

/**
 * Apply the v2 merge rule for a single incoming item against the
 * current item list and return a new list. Pure function — the input
 * array is not mutated.
 *
 * Same-`id` replaces in place (preserving array position so the UI
 * does not jitter); otherwise the item is appended.
 */
function mergeIncoming(items: FeedItem[], incoming: FeedItem): FeedItem[] {
  const idx = items.findIndex((i) => i.id === incoming.id);
  if (idx !== -1) {
    const copy = items.slice();
    copy[idx] = incoming;
    return copy;
  }
  return [...items, incoming];
}

/**
 * Return `true` when the item has an `expiresAt` timestamp that is in
 * the past relative to the supplied `nowMs`. Items without
 * `expiresAt`, or with an unparseable value, are treated as not
 * expired (fail-open).
 */
function isExpired(item: FeedItem, nowMs: number): boolean {
  if (!item.expiresAt) {
    return false;
  }
  const expiresMs = Date.parse(item.expiresAt);
  if (Number.isNaN(expiresMs)) {
    return false;
  }
  return expiresMs <= nowMs;
}

/**
 * Sort comparator: priority DESC, then createdAt DESC. Matches the
 * ordering contract the UI expects so higher-priority and fresher
 * items sort to the top of the feed.
 */
function compareFeedItems(a: FeedItem, b: FeedItem): number {
  if (a.priority !== b.priority) {
    return b.priority - a.priority;
  }
  const aMs = Date.parse(a.createdAt);
  const bMs = Date.parse(b.createdAt);
  if (Number.isNaN(aMs) && Number.isNaN(bMs)) {
    return 0;
  }
  if (Number.isNaN(aMs)) {
    return 1;
  }
  if (Number.isNaN(bMs)) {
    return -1;
  }
  return bMs - aMs;
}

/**
 * Publish a `home_feed_updated` event to the in-process hub. Wrapped
 * in a `.catch` so a subscriber rejection never bubbles up into the
 * writer coalescing loop.
 */
function publishHomeFeedUpdated(updatedAt: string, newItemCount: number): void {
  broadcastMessage({
    type: "home_feed_updated",
    updatedAt,
    newItemCount,
  });
}