import { and, asc, desc, eq, gte, inArray, isNotNull, type SQL, sql } from 'drizzle-orm';
import { getDb } from '../db';
import { articles, type Feed, feeds, fetchCache, type NewArticle, type NewFeed } from '../db/schema';
import { runAndCountChanges } from '../db/sqlite-result';
import { FeedParseError, parseFeedFromXml, validateFeedStructure } from '../parsers/feed';
import { asyncPool, cancelRateLimitedResponse, rateLimitedFetch } from '../utils/async';
import { readResponseTextWithinLimit } from '../utils/bounded-response';
import { boundArticleText, boundFeedText } from '../utils/bounded-text';
import { DEFAULT_CONCURRENCY, DEFAULT_TIMEOUT, MAX_RESPONSE_SIZE, USER_AGENT } from '../utils/constants';
import { parseHttpUrl } from '../utils/validation';
import { canonicalizeArticleLink, partitionNewArticles } from './article-dedup';
import { buildConditionalHeaders, contentHash, isXmlUnchanged } from './feed-cache';
import { categorizeError, type ErrorCategory, policyForErrorCategory, resetAfterSuccess } from './feed-error-policy';
import { FeedUrlIndex } from './feed-url-index';

export { categorizeError, type ErrorCategory };

export class FeedExistsError extends Error {
  constructor(url: string) {
    super(`Feed already exists: ${url}`);
    this.name = 'FeedExistsError';
  }
}

/**
 * Resolve canonical Feed identity, including rows written before URL
 * canonicalization was enforced. Legacy matches are intentionally returned as
 * stored: callers may edit them through updateFeed, but lookup never rewrites
 * or merges independent Feed state.
 */
async function findFeedByCanonicalUrl(canonicalUrl: string, rawUrl?: string): Promise<Feed | undefined> {
  const db = getDb();
  const [canonical] = await db.select().from(feeds).where(eq(feeds.xmlUrl, canonicalUrl)).limit(1);
  if (canonical) return canonical;

  if (rawUrl !== undefined && rawUrl !== canonicalUrl) {
    const [raw] = await db.select().from(feeds).where(eq(feeds.xmlUrl, rawUrl)).limit(1);
    if (raw) return raw;
  }

  // Legacy fallback needs a canonical scan, but only identity columns: Feed
  // descriptions/error payloads can be large and are not needed to choose a row.
  const storedFeeds = await db.select({ id: feeds.id, xmlUrl: feeds.xmlUrl }).from(feeds).orderBy(asc(feeds.id));
  const match = storedFeeds.find((feed) => {
    const parsed = parseHttpUrl(feed.xmlUrl);
    return parsed.valid && parsed.value === canonicalUrl;
  });
  if (!match) return undefined;
  const [fullFeed] = await db.select().from(feeds).where(eq(feeds.id, match.id)).limit(1);
  return fullFeed;
}

/** Exact indexed lookups only; used after a failed insert to classify a possible writer race. */
async function findFeedByExactUrl(canonicalUrl: string, rawUrl?: string): Promise<Feed | undefined> {
  const db = getDb();
  const [canonical] = await db.select().from(feeds).where(eq(feeds.xmlUrl, canonicalUrl)).limit(1);
  if (canonical) return canonical;

  if (rawUrl !== undefined && rawUrl !== canonicalUrl) {
    const [raw] = await db.select().from(feeds).where(eq(feeds.xmlUrl, rawUrl)).limit(1);
    return raw;
  }
  return undefined;
}

export interface FetchOptions {
  force?: boolean; // Force fetch even if not modified
  timeout?: number; // Request timeout in ms
  concurrency?: number; // Number of concurrent fetches
  limit?: number; // Limit number of feeds to fetch
  signal?: AbortSignal; // Abort queued/network work during coordinated shutdown
}

export interface FetchResult {
  feed: Feed;
  status: 'success' | 'not_modified' | 'error' | 'skipped';
  newArticles: number;
  error?: string;
  deactivated?: boolean;
}

export interface FeedFetchProgress {
  current: number;
  total: number;
  feed?: Feed;
  status: string;
}

function validateFetchTimeout(timeout: number | undefined): void {
  if (timeout !== undefined && (!Number.isSafeInteger(timeout) || timeout < 0)) {
    throw new RangeError('timeout must be a non-negative safe integer');
  }
}

// Fetches for a Feed are serialized within this process. The lock covers the
// network request as well as persistence so a queued fetch observes the prior
// fetch's validators, articles, and error count. It deliberately does not
// coalesce calls: each caller gets its own completed fetch result.
const feedFetchLocks = new Map<number, Promise<void>>();

function isRequestGenerationCurrent(current: Feed, requested: Feed): boolean {
  const currentFetchedAt = current.lastFetchedAt?.getTime() ?? null;
  const requestedFetchedAt = requested.lastFetchedAt?.getTime() ?? null;
  return current.xmlUrl === requested.xmlUrl && currentFetchedAt === requestedFetchedAt;
}

function nextFetchTimestamp(current: Feed, requestStartedAt: Date): Date {
  return new Date(
    Math.max(
      Date.now(),
      requestStartedAt.getTime(),
      (current.lastFetchedAt?.getTime() ?? -1) + 1,
      (current.updatedAt?.getTime() ?? -1) + 1,
    ),
  );
}

/**
 * Fetch a single feed with smart caching
 */
export async function fetchFeed(feed: Feed, options: FetchOptions = {}): Promise<FetchResult> {
  options.signal?.throwIfAborted();
  const previous = feedFetchLocks.get(feed.id) ?? Promise.resolve();
  let releaseLock: (() => void) | undefined;
  const current = new Promise<void>((resolve) => {
    releaseLock = resolve;
  });
  feedFetchLocks.set(feed.id, current);

  try {
    await previous;
    options.signal?.throwIfAborted();
    return await fetchFeedSerialized(feed, options);
  } finally {
    releaseLock?.();
    if (feedFetchLocks.get(feed.id) === current) {
      feedFetchLocks.delete(feed.id);
    }
  }
}

async function fetchFeedSerialized(requestedFeed: Feed, options: FetchOptions): Promise<FetchResult> {
  const db = getDb();
  validateFetchTimeout(options.timeout);
  const timeout = options.timeout ?? DEFAULT_TIMEOUT;
  const now = new Date();

  // A caller may have captured a Feed before a queued fetch completed. Reload
  // it after acquiring the per-Feed lock so conditional headers, metadata, and
  // error progression are all based on the latest committed state.
  const [storedFeed] = await db.select().from(feeds).where(eq(feeds.id, requestedFeed.id)).limit(1);
  if (!storedFeed) {
    return { feed: requestedFeed, status: 'skipped', newArticles: 0 };
  }
  const feed = storedFeed;

  // Check cache first (local DB read — outside the network try/catch so a
  // SQLite error here is not recorded as a feed fetch error).
  const cached = await db.select().from(fetchCache).where(eq(fetchCache.url, feed.xmlUrl)).limit(1);

  // Phase 1: network + parse. Only failures here count as feed fetch errors
  // (FETCH-A2). The DB writes that follow are kept OUT of this try so a local
  // SQLite failure (locked DB, constraint violation) is not misattributed to
  // the feed and does not increment errorCount toward auto-deactivation.
  //
  // The try resolves to a discriminated outcome describing what to persist;
  // the actual DB writes happen after the try for every branch (304,
  // unchanged-200, and changed-200) so none of them can be caught and
  // recorded as a feed fetch error.
  type FetchOutcome =
    | {
        kind: 'not_modified';
        feedEtag?: string | null;
        feedLastModified?: string | null;
        cacheEtag?: string | null;
        cacheLastModified?: string | null;
        contentHash?: string | null;
        statusCode: number;
      }
    | {
        kind: 'success';
        xml: string;
        parsed: Awaited<ReturnType<typeof parseFeedFromXml>>;
        etag?: string;
        lastModified?: string;
        hash: string;
        statusCode: number;
      };

  let outcome: FetchOutcome;

  try {
    const response = await rateLimitedFetch(
      feed.xmlUrl,
      {
        headers: {
          'User-Agent': USER_AGENT,
          Accept: 'application/rss+xml, application/atom+xml, application/xml, text/xml, */*',
          ...buildConditionalHeaders(feed, !!options.force),
        },
      },
      { timeoutMs: timeout, signal: options.signal },
    );

    // Handle 304 Not Modified — no body to read/parse.
    if (response.status === 304) {
      await cancelRateLimitedResponse(response);
      const responseEtag = response.headers.get('etag') || cached[0]?.etag || feed.etag;
      const responseLastModified =
        response.headers.get('last-modified') || cached[0]?.lastModified || feed.lastModified;
      outcome = {
        kind: 'not_modified',
        feedEtag: responseEtag,
        feedLastModified: responseLastModified,
        cacheEtag: responseEtag,
        cacheLastModified: responseLastModified,
        contentHash: cached[0]?.contentHash,
        statusCode: response.status,
      };
    } else {
      if (!response.ok) {
        await cancelRateLimitedResponse(response);
        throw new Error(`HTTP ${response.status}: ${response.statusText}`);
      }

      const xml = await readResponseTextWithinLimit(response, {
        maxBytes: MAX_RESPONSE_SIZE,
        label: 'Feed response',
      });
      const etag = response.headers.get('etag') || undefined;
      const lastModified = response.headers.get('last-modified') || undefined;

      // Validate feed structure
      const validation = validateFeedStructure(xml);
      if (!validation.valid) {
        throw new FeedParseError(validation.error ?? 'Feed validation failed');
      }

      const hash = contentHash(xml);

      // Layer-2 short-circuit: server returned 200 but body is identical to last fetch.
      if (!options.force && isXmlUnchanged(xml, cached[0]?.contentHash)) {
        outcome = {
          kind: 'not_modified',
          // ETag/Last-Modified semantic: prefer keeping the previous validator
          // when the server omits one this time, matching the success path
          // below — never silently drop a still-valid validator. The feed row
          // falls back to its own stored validator; the cache row to its own.
          feedEtag: etag ?? feed.etag,
          feedLastModified: lastModified ?? feed.lastModified,
          cacheEtag: etag ?? cached[0]?.etag,
          cacheLastModified: lastModified ?? cached[0]?.lastModified,
          contentHash: hash,
          statusCode: response.status,
        };
      } else {
        // Parse feed
        const parsed = await parseFeedFromXml(xml, { etag, lastModified, feedUrl: feed.xmlUrl });
        outcome = { kind: 'success', xml, parsed, etag, lastModified, hash, statusCode: response.status };
      }
    }
  } catch (error) {
    options.signal?.throwIfAborted();
    const errorMessage = error instanceof Error ? error.message : String(error);
    const category = error instanceof FeedParseError ? error.category : categorizeError(errorMessage);
    const policy = policyForErrorCategory(category);
    const categoryGroup = category === 'permanent' ? 'permanent' : 'recoverable';
    const failed = db.transaction(
      (tx) => {
        const current = tx.select().from(feeds).where(eq(feeds.id, feed.id)).get();
        if (!current || !isRequestGenerationCurrent(current, feed)) return undefined;

        const nextErrorCount = sql<number>`case
          when ${feeds.errorCategory} = ${categoryGroup} then coalesce(${feeds.errorCount}, 0) + 1
          else 1
        end`;
        const updatedFeed = tx
          .update(feeds)
          .set({
            fetchError: errorMessage,
            errorCategory: categoryGroup,
            // Compute the streak transition and deactivation from the current
            // row in one UPDATE. Separate CLI processes can finish failed
            // requests concurrently; a read-then-write increment would lose
            // one of those failures.
            errorCount: nextErrorCount,
            isActive: sql`case
              when ${nextErrorCount} >= ${policy.threshold} then 0
              else ${feeds.isActive}
            end`,
            updatedAt: new Date(Math.max(now.getTime(), (current.updatedAt?.getTime() ?? -1) + 1)),
          })
          .where(eq(feeds.id, current.id))
          .returning({ errorCount: feeds.errorCount, isActive: feeds.isActive })
          .get();
        return { current, updatedFeed };
      },
      { behavior: 'immediate' },
    );

    if (!failed?.updatedFeed) {
      return { feed, status: 'skipped', newArticles: 0 };
    }
    const { current, updatedFeed } = failed;
    const newErrorCount = updatedFeed.errorCount ?? (current.errorCount ?? 0) + 1;
    const deactivated = updatedFeed ? !updatedFeed.isActive && newErrorCount >= policy.threshold : false;

    return {
      feed: current,
      status: 'error',
      newArticles: 0,
      error: deactivated ? `${errorMessage} (deactivated after ${newErrorCount} consecutive errors)` : errorMessage,
      deactivated,
    };
  }

  // Phase 2: persist. Errors here (local SQLite) propagate to the caller as-is
  // rather than being recorded as a feed fetch error (FETCH-A2). This covers
  // the 304 / unchanged-200 branches too, which is why their DB writes live
  // here outside the try/catch rather than inline above.
  options.signal?.throwIfAborted();

  if (outcome.kind === 'not_modified') {
    const persistedFeed = db.transaction(
      (tx) => {
        const current = tx.select().from(feeds).where(eq(feeds.id, feed.id)).get();
        if (!current || !isRequestGenerationCurrent(current, feed)) return undefined;
        const persistedAt = nextFetchTimestamp(current, now);

        tx.update(feeds)
          .set({
            lastFetchedAt: persistedAt,
            etag: outcome.feedEtag,
            lastModified: outcome.feedLastModified,
            ...resetAfterSuccess(),
            errorCategory: null,
            updatedAt: persistedAt,
          })
          .where(eq(feeds.id, current.id))
          .run();

        tx.insert(fetchCache)
          .values({
            url: current.xmlUrl,
            etag: outcome.cacheEtag,
            lastModified: outcome.cacheLastModified,
            contentHash: outcome.contentHash,
            statusCode: outcome.statusCode,
            fetchedAt: persistedAt,
          })
          .onConflictDoUpdate({
            target: fetchCache.url,
            set: {
              etag: outcome.cacheEtag,
              lastModified: outcome.cacheLastModified,
              contentHash: outcome.contentHash,
              statusCode: outcome.statusCode,
              fetchedAt: persistedAt,
            },
          })
          .run();
        return current;
      },
      { behavior: 'immediate' },
    );

    if (!persistedFeed) {
      return { feed, status: 'skipped', newArticles: 0 };
    }

    return {
      feed: persistedFeed,
      status: 'not_modified',
      newArticles: 0,
    };
  }

  const { parsed, etag, lastModified, hash, statusCode } = outcome;
  // Defense in depth for alternate/programmatic parser callers: repeat all
  // publisher-controlled bounds and Article URL canonicalization at the exact
  // persistence boundary, before identity lookup and JSON serialization.
  const boundedFeed = boundFeedText(parsed);
  const boundedArticles = parsed.articles.map((candidate) => {
    const bounded = boundArticleText(candidate);
    return { ...bounded, link: canonicalizeArticleLink(bounded.link) };
  });

  const persisted = db.transaction(
    (tx) => {
      // The request can take arbitrarily long. Re-read under the write
      // transaction so a concurrent `feed update` is never overwritten by the
      // old snapshot we fetched with. A URL change means this response and its
      // validators belong to a different resource, so discard it entirely.
      const current = tx.select().from(feeds).where(eq(feeds.id, feed.id)).get();
      if (!current || !isRequestGenerationCurrent(current, feed)) return undefined;
      const persistedAt = nextFetchTimestamp(current, now);

      // Feed freshness, cache validators, identity lookup, and Article writes
      // commit together. If an Article insert fails, a subsequent request cannot
      // receive a 304 based on validators/content hash that were never backed by
      // a complete article persistence.
      tx.update(feeds)
        .set({
          // Preserve manual title overrides set via `feed update --title`, even
          // if that change happened while the network request was in flight.
          title: current.titleOverridden ? current.title : boundedFeed.title || current.title,
          htmlUrl: boundedFeed.link || current.htmlUrl,
          description: boundedFeed.description || current.description,
          lastFetchedAt: persistedAt,
          etag: etag ?? current.etag,
          lastModified: lastModified ?? current.lastModified,
          ...resetAfterSuccess(),
          errorCategory: null,
          updatedAt: persistedAt,
        })
        .where(eq(feeds.id, current.id))
        .run();

      tx.insert(fetchCache)
        .values({
          url: current.xmlUrl,
          etag: etag ?? cached[0]?.etag,
          lastModified: lastModified ?? cached[0]?.lastModified,
          contentHash: hash,
          statusCode,
          fetchedAt: persistedAt,
        })
        .onConflictDoUpdate({
          target: fetchCache.url,
          set: {
            etag: etag ?? cached[0]?.etag,
            lastModified: lastModified ?? cached[0]?.lastModified,
            contentHash: hash,
            statusCode,
            fetchedAt: persistedAt,
          },
        })
        .run();

      // Query only indexed identities represented in this payload. Keep chunks
      // comfortably below SQLite's conservative 999-variable limit (each query
      // also binds feed_id), while retaining both columns for GUID-then-link
      // first-wins semantics in partitionNewArticles.
      const incomingGuids = [...new Set(boundedArticles.flatMap((article) => (article.guid ? [article.guid] : [])))];
      const incomingLinks = [...new Set(boundedArticles.flatMap((article) => (article.link ? [article.link] : [])))];
      const existing: Array<{ guid: string | null; link: string | null }> = [];
      const IDENTITY_LOOKUP_CHUNK_SIZE = 400;
      for (let i = 0; i < incomingGuids.length; i += IDENTITY_LOOKUP_CHUNK_SIZE) {
        existing.push(
          ...tx
            .select({ guid: articles.guid, link: articles.link })
            .from(articles)
            .where(
              and(
                eq(articles.feedId, current.id),
                inArray(articles.guid, incomingGuids.slice(i, i + IDENTITY_LOOKUP_CHUNK_SIZE)),
              ),
            )
            .all(),
        );
      }
      for (let i = 0; i < incomingLinks.length; i += IDENTITY_LOOKUP_CHUNK_SIZE) {
        existing.push(
          ...tx
            .select({ guid: articles.guid, link: articles.link })
            .from(articles)
            .where(
              and(
                eq(articles.feedId, current.id),
                inArray(articles.link, incomingLinks.slice(i, i + IDENTITY_LOOKUP_CHUNK_SIZE)),
              ),
            )
            .all(),
        );
      }

      const newArticlesToInsert: NewArticle[] = partitionNewArticles(boundedArticles, existing).map((candidate) => {
        const article = boundArticleText(candidate);
        return {
          feedId: current.id,
          guid: article.guid,
          title: article.title,
          link: article.link,
          description: article.description,
          content: article.content,
          author: article.author,
          categories: article.categories ? JSON.stringify(article.categories) : null,
          publishedAt: article.publishedAt,
          createdAt: now,
          updatedAt: now,
        };
      });

      // Batch insert new articles, chunked to stay within SQLite variable limits.
      const BATCH_SIZE = 500;
      if (newArticlesToInsert.length > 0) {
        for (let i = 0; i < newArticlesToInsert.length; i += BATCH_SIZE) {
          const chunk = newArticlesToInsert.slice(i, i + BATCH_SIZE);
          tx.insert(articles).values(chunk).run();
        }
      }
      return { feed: current, newArticles: newArticlesToInsert.length };
    },
    { behavior: 'immediate' },
  );

  options.signal?.throwIfAborted();
  if (!persisted) {
    return { feed, status: 'skipped', newArticles: 0 };
  }

  return {
    feed: persisted.feed,
    status: 'success',
    newArticles: persisted.newArticles,
  };
}

/**
 * Fetch all feeds with true concurrent pool processing.
 * Maintains exactly N active fetches at all times, starting a new fetch
 * as soon as any slot frees up (unlike batch processing which waits for
 * all items in a batch before starting the next).
 */
export async function fetchAllFeeds(
  options: FetchOptions = {},
  onProgress?: (progress: FeedFetchProgress) => void,
): Promise<FetchResult[]> {
  options.signal?.throwIfAborted();
  const db = getDb();
  const concurrency = options.concurrency ?? DEFAULT_CONCURRENCY;
  if (!Number.isSafeInteger(concurrency) || concurrency <= 0) {
    throw new RangeError('concurrency must be a finite positive integer');
  }
  if (options.limit !== undefined && (!Number.isSafeInteger(options.limit) || options.limit < 0)) {
    throw new RangeError('limit must be a non-negative safe integer');
  }
  validateFetchTimeout(options.timeout);

  // Get all active feeds. When limited, fetch the least-recently-fetched feeds
  // first (NULLs — never-fetched — first) so repeated `--limit N` runs rotate
  // fairly instead of always starving newer feeds (FETCH-A3).
  const allFeeds =
    options.limit !== undefined
      ? await db
          .select()
          .from(feeds)
          .where(eq(feeds.isActive, true))
          .orderBy(sql`${feeds.lastFetchedAt} asc nulls first`, asc(feeds.id))
          .limit(options.limit)
      : await db.select().from(feeds).where(eq(feeds.isActive, true));
  const total = allFeeds.length;

  // Track progress counter atomically
  let completed = 0;

  return asyncPool(allFeeds, concurrency, async (feed) => {
    const result = await fetchFeed(feed, options);
    completed++;
    onProgress?.({
      current: completed,
      total,
      feed,
      status: result.status,
    });
    return result;
  });
}

/**
 * Add a new feed
 */
export type AddFeedResult = Feed & {
  /** Outcome of the initial fetch performed by addFeed, when fetchNow !== false. */
  initialFetchStatus?: FetchResult['status'];
  /** Error message from the initial fetch, if it failed. */
  initialFetchError?: string;
};

export interface AddFeedOptions {
  title?: string;
  category?: string;
  fetchNow?: boolean;
  titleOverridden?: boolean;
  signal?: AbortSignal;
}

async function insertFeed(canonicalXmlUrl: string, rawXmlUrl: string, options: AddFeedOptions): Promise<Feed> {
  const db = getDb();
  const now = new Date();
  const boundedOptions = boundFeedText({ title: options.title, category: options.category });
  const newFeed: NewFeed = {
    title: boundedOptions.title || canonicalXmlUrl,
    titleOverridden: options.titleOverridden ?? options.title !== undefined,
    xmlUrl: canonicalXmlUrl,
    category: boundedOptions.category,
    createdAt: now,
    updatedAt: now,
  };

  try {
    const [feed] = await db.insert(feeds).values(newFeed).returning();
    if (!feed) throw new Error('Failed to create feed');
    return feed;
  } catch (error) {
    // Only indexed equality checks are allowed here. A full legacy scan could
    // misclassify an unrelated persistence failure as a duplicate.
    let raceWinner: Feed | undefined;
    try {
      raceWinner = await findFeedByExactUrl(canonicalXmlUrl, rawXmlUrl);
    } catch {
      // A failed diagnostic lookup must not replace the original insert error.
      throw error;
    }
    if (raceWinner) {
      throw new FeedExistsError(canonicalXmlUrl);
    }
    throw error;
  }
}

async function finishAddedFeed(feed: Feed, options: AddFeedOptions): Promise<AddFeedResult> {
  if (options.fetchNow !== false) {
    const initial = await fetchFeed(feed, { force: true, signal: options.signal });
    return { ...feed, initialFetchStatus: initial.status, initialFetchError: initial.error };
  }
  return feed;
}

export async function addFeed(xmlUrl: string, options: AddFeedOptions = {}): Promise<AddFeedResult> {
  const parsedUrl = parseHttpUrl(xmlUrl);
  if (!parsedUrl.valid) throw new Error(parsedUrl.error);
  const canonicalXmlUrl = parsedUrl.value;

  // Check if feed already exists
  const existing = await findFeedByCanonicalUrl(canonicalXmlUrl, xmlUrl);
  if (existing) {
    throw new FeedExistsError(canonicalXmlUrl);
  }

  const feed = await insertFeed(canonicalXmlUrl, xmlUrl, options);
  return finishAddedFeed(feed, options);
}

export interface FeedAddSession {
  /** Reserve a URL identity without writing, for bulk-import previews. */
  claim(xmlUrl: string): boolean;
  /** Add using this session's one-time existing-Feed identity index. */
  addFeed(xmlUrl: string, options?: AddFeedOptions): Promise<AddFeedResult>;
}

/**
 * Create a bulk-add session with one O(existing) URL identity scan. All later
 * duplicate checks and successful-insert updates are O(1) in memory.
 */
export async function createFeedAddSession(): Promise<FeedAddSession> {
  // Duplicate identity needs only the narrow indexed fields; avoid retaining
  // every Feed's descriptive/error metadata for a potentially large import.
  const existingFeeds = await getDb().select({ id: feeds.id, xmlUrl: feeds.xmlUrl }).from(feeds).orderBy(asc(feeds.id));
  const index = new FeedUrlIndex(existingFeeds);

  return {
    claim(xmlUrl: string): boolean {
      const parsed = parseHttpUrl(xmlUrl);
      if (!parsed.valid) throw new Error(parsed.error);
      return index.claim(parsed.value, xmlUrl);
    },

    async addFeed(xmlUrl: string, options: AddFeedOptions = {}): Promise<AddFeedResult> {
      const parsed = parseHttpUrl(xmlUrl);
      if (!parsed.valid) throw new Error(parsed.error);
      const canonicalXmlUrl = parsed.value;
      if (!index.claim(canonicalXmlUrl, xmlUrl)) {
        throw new FeedExistsError(canonicalXmlUrl);
      }

      try {
        const feed = await insertFeed(canonicalXmlUrl, xmlUrl, options);
        index.record(feed, canonicalXmlUrl);
        return await finishAddedFeed(feed, options);
      } catch (error) {
        index.release(canonicalXmlUrl);
        throw error;
      }
    },
  };
}

/**
 * Remove a feed
 */
export async function removeFeed(feedId: number): Promise<void> {
  const db = getDb();
  await db.delete(feeds).where(eq(feeds.id, feedId));
}

/**
 * Get feed by ID
 */
export async function getFeed(feedId: number): Promise<Feed | undefined> {
  const db = getDb();
  const [feed] = await db.select().from(feeds).where(eq(feeds.id, feedId)).limit(1);
  return feed;
}

/**
 * Get feed by URL
 */
export async function getFeedByUrl(url: string): Promise<Feed | undefined> {
  const parsed = parseHttpUrl(url);
  if (!parsed.valid) throw new Error(parsed.error);
  return findFeedByCanonicalUrl(parsed.value, url);
}

/**
 * List all feeds
 */
export async function listFeeds(
  options: { category?: string; active?: boolean; hasErrors?: boolean } = {},
): Promise<Feed[]> {
  const db = getDb();

  const conditions: SQL[] = [];

  if (options.category) {
    conditions.push(eq(feeds.category, options.category));
  }

  if (options.active !== undefined) {
    conditions.push(eq(feeds.isActive, options.active));
  }

  if (options.hasErrors) {
    conditions.push(isNotNull(feeds.fetchError));
  }

  const query =
    conditions.length > 0
      ? db
          .select()
          .from(feeds)
          .where(and(...conditions))
      : db.select().from(feeds);

  return query.orderBy(desc(feeds.createdAt), desc(feeds.id));
}

/**
 * Update feed
 */
export async function updateFeed(
  feedId: number,
  updates: Partial<Pick<Feed, 'title' | 'xmlUrl' | 'category' | 'isActive'>>,
): Promise<Feed | undefined> {
  const db = getDb();
  const parsedUrl = updates.xmlUrl === undefined ? undefined : parseHttpUrl(updates.xmlUrl);
  if (parsedUrl && !parsedUrl.valid) throw new Error(parsedUrl.error);
  if (parsedUrl?.valid) {
    const existing = await findFeedByCanonicalUrl(parsedUrl.value, updates.xmlUrl);
    if (existing && existing.id !== feedId) throw new FeedExistsError(parsedUrl.value);
  }

  // Mark title as manually overridden so refetches don't clobber it
  const persisted: Partial<Feed> & { updatedAt: Date } = { ...updates, updatedAt: new Date() };
  const boundedUpdates = boundFeedText({ title: updates.title, category: updates.category });
  if (updates.title !== undefined) persisted.title = boundedUpdates.title;
  if (updates.category !== undefined) persisted.category = boundedUpdates.category;
  if (parsedUrl?.valid) persisted.xmlUrl = parsedUrl.value;
  if (updates.title !== undefined) {
    persisted.titleOverridden = true;
  }

  try {
    return db.transaction(
      (tx) => {
        const current = tx.select().from(feeds).where(eq(feeds.id, feedId)).get();
        if (!current) return undefined;

        if (persisted.xmlUrl !== undefined && persisted.xmlUrl !== current.xmlUrl) {
          persisted.etag = null;
          persisted.lastModified = null;
          persisted.lastFetchedAt = null;
          persisted.fetchError = null;
          persisted.errorCount = 0;
          persisted.errorCategory = null;
          tx.delete(fetchCache).where(eq(fetchCache.url, current.xmlUrl)).run();
        }

        return tx.update(feeds).set(persisted).where(eq(feeds.id, feedId)).returning().get();
      },
      { behavior: 'immediate' },
    );
  } catch (error) {
    if (parsedUrl?.valid) {
      const conflict = await findFeedByCanonicalUrl(parsedUrl.value, updates.xmlUrl);
      if (conflict && conflict.id !== feedId) throw new FeedExistsError(parsedUrl.value);
    }
    throw error;
  }
}

/**
 * Get feed categories
 */
export async function getCategories(): Promise<string[]> {
  const db = getDb();

  const result = await db
    .selectDistinct({ category: feeds.category })
    .from(feeds)
    .where(isNotNull(feeds.category))
    .orderBy(asc(feeds.category));

  return result.map((r) => r.category).filter((c): c is string => c !== null);
}

/**
 * Remove feeds with errors
 */
export async function removeBrokenFeeds(
  options: {
    dryRun?: boolean;
    errorCount?: number; // Minimum error count
  } = {},
): Promise<{ removed: number; matching?: number; preview?: Feed[] }> {
  if (options.errorCount !== undefined && (!Number.isSafeInteger(options.errorCount) || options.errorCount < 0)) {
    throw new RangeError('errorCount must be a non-negative safe integer');
  }
  const db = getDb();

  // Build conditions
  const conditions: SQL[] = [isNotNull(feeds.fetchError)];

  if (options.errorCount !== undefined) {
    conditions.push(gte(feeds.errorCount, options.errorCount));
  }

  // For dry run, fetch preview
  if (options.dryRun) {
    const [countRows, preview] = await Promise.all([
      db
        .select({ count: sql<number>`count(*)` })
        .from(feeds)
        .where(and(...conditions)),
      db
        .select()
        .from(feeds)
        .where(and(...conditions))
        .orderBy(asc(feeds.id))
        .limit(100),
    ]);

    return { removed: 0, matching: countRows[0]?.count ?? 0, preview };
  }

  // Count top-level Feed rows, not cascading Article/FTS trigger writes.
  const removed = runAndCountChanges(db, () =>
    db
      .delete(feeds)
      .where(and(...conditions))
      .run(),
  );
  return { removed };
}