import * as cheerio from 'cheerio';
import { and, asc, desc, eq, gt, inArray, isNull, lt, lte, or, type SQL, sql } from 'drizzle-orm';
import { getDb } from '../db';
import { type Article, articles, maintenanceCursors } from '../db/schema';
import { allowPrivateArticleNetwork } from '../utils/article-network-policy';
import { asyncPool, cancelRateLimitedResponse } from '../utils/async';
import { readResponseTextWithinLimit } from '../utils/bounded-response';
import { truncateUtf16 } from '../utils/bounded-text';
import {
  CONTENT_SELECTORS,
  DEFAULT_ARTICLE_LIMIT,
  DEFAULT_CONCURRENCY,
  DEFAULT_TIMEOUT,
  MAX_CONTENT_SIZE,
  MAX_RESPONSE_SIZE,
  MIN_CONTENT_LENGTH,
  USER_AGENT,
} from '../utils/constants';
import { hasRenderableContent, htmlToMarkdown } from '../utils/html';
import { fetchArticleRequest } from './article-request';
import { type CheckUrlResult, checkArticleUrl, tryFixBrokenArticleUrl } from './broken-link-repair';
import { decideBrokenLinkAction, isDefinitivelyBroken } from './broken-link-strategy';

/**
 * Failed extraction attempts are retried by unattended batch work only after
 * this cooldown. `contentFetchedAt` therefore means "last extraction attempt"
 * rather than proving that `content` came from the extractor; RSS/Atom content
 * remains usable without that timestamp.
 */
export const ARTICLE_CONTENT_RETRY_DELAY_MS = 6 * 60 * 60 * 1000;

export interface ArticleFetchOptions {
  timeout?: number;
  userAgent?: string;
  concurrency?: number;
  /** Explicitly allow private/LAN Article destinations (cloud metadata remains blocked). */
  allowPrivateNetwork?: boolean;
  signal?: AbortSignal;
}

export interface ArticleFetchResult {
  article: Article;
  status: 'success' | 'error' | 'skipped';
  /** True only when this operation won its conditional database write. */
  persisted: boolean;
  /** True when `article` was read from the current database state and can refresh a UI view. */
  current: boolean;
  error?: string;
}

export interface ArticleContentFetchResult {
  content?: string;
  title?: string;
  error?: string;
  /** HTTP status returned by the direct content GET; absent for failures before a response. */
  statusCode?: number;
}

type ArticleFetchBehavior = {
  fixBroken?: boolean;
  skipBroken?: boolean;
  flagBroken?: boolean;
};

type CompleteArticleFetchOptions = ArticleFetchOptions & ArticleFetchBehavior;

interface ArticlePersistenceResult {
  article: Article;
  persisted: boolean;
  current: boolean;
}

interface ArticleUpdate {
  content?: string;
  contentFetchedAt: Date;
  isBroken?: boolean;
  brokenReason?: string | null;
  linkFixed?: string | null;
  updatedAt: Date;
}

interface SharedArticleFetch {
  controller: AbortController;
  promise: Promise<ArticleFetchResult>;
  subscribers: number;
  settled: boolean;
}

class ArticleOperationTimeoutError extends Error {
  constructor(timeout: number) {
    super(`Article content fetch timed out after ${timeout}ms`);
    this.name = 'ArticleOperationTimeoutError';
  }
}

const inFlightArticleFetches = new Map<string, SharedArticleFetch>();

function resolveTimeout(timeout: number | undefined): number {
  return timeout ?? DEFAULT_TIMEOUT;
}

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

function validateArticleConcurrency(concurrency: number | undefined): void {
  if (concurrency !== undefined && (!Number.isSafeInteger(concurrency) || concurrency <= 0)) {
    throw new RangeError('concurrency must be a finite positive integer');
  }
}

function isOperationTimeout(signal: AbortSignal | undefined): boolean {
  return signal?.aborted === true && signal.reason instanceof ArticleOperationTimeoutError;
}

function errorMessage(error: unknown): string {
  return error instanceof Error ? error.message : String(error);
}

function abortError(signal: AbortSignal): Error {
  return signal.reason instanceof Error ? signal.reason : new Error('Article fetch aborted');
}

/** Throw only for caller cancellation. An internal timeout becomes a recorded retryable attempt. */
function throwIfCallerCancelled(signal: AbortSignal | undefined): void {
  if (!signal?.aborted || isOperationTimeout(signal)) return;
  throw abortError(signal);
}

function createOperationDeadline(
  callerSignal: AbortSignal | undefined,
  timeout: number,
): {
  signal: AbortSignal;
  cleanup: () => void;
} {
  const controller = new AbortController();
  const onCallerAbort = () => controller.abort(callerSignal?.reason);
  if (callerSignal?.aborted) {
    controller.abort(callerSignal.reason);
  } else {
    callerSignal?.addEventListener('abort', onCallerAbort, { once: true });
  }
  const timer = setTimeout(() => controller.abort(new ArticleOperationTimeoutError(timeout)), timeout);

  return {
    signal: controller.signal,
    cleanup: () => {
      clearTimeout(timer);
      callerSignal?.removeEventListener('abort', onCallerAbort);
    },
  };
}

/** Fetch article content from one URL without changing Article state. */
export async function fetchArticleContent(
  url: string,
  options: ArticleFetchOptions = {},
): Promise<ArticleContentFetchResult> {
  validateArticleTimeout(options.timeout);
  const timeout = resolveTimeout(options.timeout);
  let statusCode: number | undefined;

  try {
    throwIfCallerCancelled(options.signal);
    if (isOperationTimeout(options.signal)) {
      return { error: errorMessage(options.signal?.reason) };
    }

    const { response } = await fetchArticleRequest(
      url,
      {
        headers: {
          'User-Agent': options.userAgent || USER_AGENT,
          Accept: 'text/html,application/xhtml+xml,text/plain',
        },
      },
      {
        timeoutMs: timeout,
        signal: options.signal,
        allowPrivateNetwork: options.allowPrivateNetwork,
      },
    );
    statusCode = response.status;

    if (!response.ok) {
      await cancelRateLimitedResponse(response);
      return { error: `HTTP ${response.status}: ${response.statusText}`, statusCode };
    }

    const html = await readResponseTextWithinLimit(response, {
      maxBytes: MAX_RESPONSE_SIZE,
      label: 'Article response',
    });

    const $ = cheerio.load(html);
    $('script, style, nav, header, footer, aside, .ads, .advertisement, .comments').remove();

    let content = '';
    // Evaluate each matched element on its OWN text so a near-empty first match
    // can't ride on the combined length of later matches: pick the first
    // individual element whose own text passes the threshold.
    outer: for (const selector of CONTENT_SELECTORS) {
      const matches = $(selector);
      for (const node of matches.toArray()) {
        const element = $(node);
        if (element.text().trim().length > MIN_CONTENT_LENGTH) {
          const rawHtml = element.html()?.trim() || '';
          content = rawHtml ? htmlToMarkdown(rawHtml) : '';
          if (content) break outer;
        }
      }
    }

    if (!content) {
      const rawHtml = $('body').html()?.trim() || '';
      content = rawHtml ? htmlToMarkdown(rawHtml) : '';
    }

    const trimmedContent = content.trim();
    if (!trimmedContent) {
      return { error: 'No article content found', statusCode };
    }

    const title =
      $('title').text().trim() || $('h1').first().text().trim() || $('meta[property="og:title"]').attr('content');

    return {
      content: truncateUtf16(trimmedContent, MAX_CONTENT_SIZE),
      title,
      statusCode,
    };
  } catch (error) {
    if (isOperationTimeout(options.signal)) {
      return { error: errorMessage(options.signal?.reason), statusCode };
    }
    throwIfCallerCancelled(options.signal);
    return { error: errorMessage(error), statusCode };
  }
}

function checkFromDirectFetch(direct: ArticleContentFetchResult, error: string): CheckUrlResult {
  return {
    accessible: direct.statusCode !== undefined && direct.statusCode >= 200 && direct.statusCode < 300,
    statusCode: direct.statusCode,
    error,
  };
}

function articleVersionCondition(article: Article) {
  return and(
    eq(articles.id, article.id),
    sql<boolean>`${articles.link} IS ${article.link}`,
    sql<boolean>`${articles.content} IS ${article.content}`,
    sql<boolean>`${articles.isBroken} IS ${article.isBroken}`,
    sql<boolean>`${articles.brokenReason} IS ${article.brokenReason}`,
    sql<boolean>`${articles.linkFixed} IS ${article.linkFixed}`,
    article.updatedAt ? eq(articles.updatedAt, article.updatedAt) : isNull(articles.updatedAt),
    article.contentFetchedAt
      ? eq(articles.contentFetchedAt, article.contentFetchedAt)
      : isNull(articles.contentFetchedAt),
  );
}

/**
 * Persist one extraction outcome optimistically. A newer Article content/url
 * state wins over stale work; harmless metadata changes get one retry against
 * the newer row so a read/star click cannot discard a completed extraction.
 */
async function persistArticleOutcome(
  snapshot: Article,
  signal: AbortSignal,
  changesFor: (current: Article, now: Date) => ArticleUpdate,
): Promise<ArticlePersistenceResult> {
  const db = getDb();
  let current = snapshot;

  for (let attempt = 0; attempt < 2; attempt++) {
    if (hasRenderableContent(current.content) || current.link !== snapshot.link) {
      return { article: current, persisted: false, current: current !== snapshot };
    }

    // Bun's SQLite call begins synchronously after this point. Checking at the
    // write boundary keeps cancellation from publishing a late Article state.
    throwIfCallerCancelled(signal);
    const [updated] = await db
      .update(articles)
      .set(changesFor(current, new Date()))
      .where(articleVersionCondition(current))
      .returning();
    if (updated) return { article: updated, persisted: true, current: true };

    const [latest] = await db.select().from(articles).where(eq(articles.id, snapshot.id)).limit(1);
    if (!latest) return { article: snapshot, persisted: false, current: false };
    current = latest;
  }

  return { article: current, persisted: false, current: true };
}

function resultFromPersistence(
  persistence: ArticlePersistenceResult,
  status: ArticleFetchResult['status'],
  error?: string,
): ArticleFetchResult {
  return {
    article: persistence.article,
    status: persistence.persisted ? status : 'skipped',
    persisted: persistence.persisted,
    current: persistence.current,
    error,
  };
}

async function persistAttempt(
  article: Article,
  signal: AbortSignal,
  status: 'error' | 'skipped',
  error: string,
): Promise<ArticleFetchResult> {
  const persistence = await persistArticleOutcome(article, signal, (_current, now) => ({
    contentFetchedAt: now,
    updatedAt: now,
  }));
  return resultFromPersistence(persistence, status, error);
}

async function persistBrokenAttempt(
  article: Article,
  signal: AbortSignal,
  reason: string,
): Promise<ArticleFetchResult> {
  const persistence = await persistArticleOutcome(article, signal, (_current, now) => ({
    contentFetchedAt: now,
    isBroken: true,
    brokenReason: reason,
    updatedAt: now,
  }));
  return resultFromPersistence(persistence, 'error', reason);
}

async function persistSuccess(
  article: Article,
  signal: AbortSignal,
  content: string,
  resolvedUrl: string,
): Promise<ArticleFetchResult> {
  const persistence = await persistArticleOutcome(article, signal, (current, now) => ({
    content,
    contentFetchedAt: now,
    isBroken: false,
    brokenReason: null,
    // Preserve an earlier repair when the original URL succeeded this time;
    // a repaired URL from this operation replaces it deliberately.
    linkFixed: resolvedUrl !== current.link ? resolvedUrl : (current.linkFixed ?? null),
    updatedAt: now,
  }));
  return resultFromPersistence(persistence, 'success');
}

async function fetchArticleOnce(article: Article, options: CompleteArticleFetchOptions): Promise<ArticleFetchResult> {
  const deadline = createOperationDeadline(options.signal, resolveTimeout(options.timeout));
  const signal = deadline.signal;

  try {
    throwIfCallerCancelled(signal);
    if (hasRenderableContent(article.content)) {
      return { article, status: 'skipped', persisted: false, current: false };
    }

    const direct = await fetchArticleContent(article.link, { ...options, signal });
    if (isOperationTimeout(signal)) {
      return persistAttempt(article, signal, 'error', errorMessage(signal.reason));
    }
    throwIfCallerCancelled(signal);
    if (!direct.error && direct.content) {
      return persistSuccess(article, signal, direct.content, article.link);
    }

    const directError = direct.error ?? 'No article content found';
    const directCheck = checkFromDirectFetch(direct, directError);
    const directIsDefinitivelyBroken = isDefinitivelyBroken(directCheck);
    const originalDefinitiveDecision = directIsDefinitivelyBroken
      ? decideBrokenLinkAction(article.link, directCheck, undefined, options)
      : undefined;
    // A direct 404/410 already proves original-link brokenness. Re-probing the
    // same URL can only contradict the authoritative GET and adds needless I/O.
    // Other failures still use the reachability probe when skip/repair behavior
    // needs it, but that probe is never allowed to authorize Broken state.
    const needsReachabilityProbe = !directIsDefinitivelyBroken;
    const check = needsReachabilityProbe ? await checkArticleUrl(article.link, { ...options, signal }) : directCheck;
    if (isOperationTimeout(signal)) {
      return persistAttempt(article, signal, 'error', errorMessage(signal.reason));
    }
    throwIfCallerCancelled(signal);

    const fix =
      !check.accessible && options.fixBroken
        ? await tryFixBrokenArticleUrl(article.link, { ...options, signal })
        : undefined;
    if (isOperationTimeout(signal)) {
      return persistAttempt(article, signal, 'error', errorMessage(signal.reason));
    }
    throwIfCallerCancelled(signal);

    const decision = decideBrokenLinkAction(
      article.link,
      check,
      fix,
      directIsDefinitivelyBroken ? options : { ...options, flagBroken: false },
    );
    if (decision.kind === 'skip') {
      // Record the attempt without claiming that an unreachable response was
      // deletion-eligible. This prevents one bad link from monopolizing every
      // default batch while keeping the Article recoverable.
      return persistAttempt(article, signal, 'skipped', decision.reason);
    }
    if (decision.kind === 'persist-broken') {
      return persistBrokenAttempt(article, signal, decision.reason);
    }
    if (decision.kind === 'fail') {
      return persistAttempt(article, signal, 'error', decision.reason);
    }

    // A successful repair gets one extraction attempt. A direct extraction
    // failure followed by an accessible probe does not get the same URL a
    // second time: reachability and extractability are distinct facts.
    if (decision.url === article.link) {
      return persistAttempt(article, signal, 'error', directError);
    }

    const repaired = await fetchArticleContent(decision.url, { ...options, signal });
    if (isOperationTimeout(signal)) {
      return persistAttempt(article, signal, 'error', errorMessage(signal.reason));
    }
    throwIfCallerCancelled(signal);
    if (!repaired.error && repaired.content) {
      return persistSuccess(article, signal, repaired.content, decision.url);
    }

    // HEAD only establishes that the candidate URL is reachable. If its
    // subsequent content GET cannot produce an article, the original direct
    // 404/410 remains authoritative and retains its flag/skip/fail policy.
    // Do not re-probe the original or replace its reason with the candidate's.
    if (originalDefinitiveDecision?.kind === 'skip') {
      return persistAttempt(article, signal, 'skipped', originalDefinitiveDecision.reason);
    }
    if (originalDefinitiveDecision?.kind === 'persist-broken') {
      return persistBrokenAttempt(article, signal, originalDefinitiveDecision.reason);
    }
    if (originalDefinitiveDecision?.kind === 'fail') {
      return persistAttempt(article, signal, 'error', originalDefinitiveDecision.reason);
    }
    return persistAttempt(article, signal, 'error', repaired.error ?? 'No article content found');
  } catch (error) {
    if (isOperationTimeout(signal)) {
      return persistAttempt(article, signal, 'error', errorMessage(signal.reason));
    }
    throw error;
  } finally {
    deadline.cleanup();
  }
}

function articleFetchKey(article: Article, options: CompleteArticleFetchOptions): string {
  return [
    article.id,
    article.link,
    article.content ?? '',
    article.updatedAt?.getTime() ?? '',
    article.contentFetchedAt?.getTime() ?? '',
    article.isBroken ? 1 : 0,
    article.brokenReason ?? '',
    article.linkFixed ?? '',
    resolveTimeout(options.timeout),
    options.userAgent ?? '',
    options.fixBroken ? 1 : 0,
    options.skipBroken ? 1 : 0,
    options.flagBroken ? 1 : 0,
    allowPrivateArticleNetwork(options.allowPrivateNetwork) ? 1 : 0,
  ].join('\u0000');
}

function joinSharedFetch(shared: SharedArticleFetch, signal: AbortSignal | undefined): Promise<ArticleFetchResult> {
  shared.subscribers++;

  return new Promise<ArticleFetchResult>((resolve, reject) => {
    let released = false;
    const release = () => {
      if (released) return;
      released = true;
      signal?.removeEventListener('abort', onAbort);
      shared.subscribers--;
      if (shared.subscribers === 0 && !shared.settled && !shared.controller.signal.aborted) {
        shared.controller.abort(signal?.reason);
      }
    };
    const onAbort = () => {
      release();
      reject(abortError(signal as AbortSignal));
    };

    if (signal?.aborted) {
      onAbort();
      return;
    }
    signal?.addEventListener('abort', onAbort, { once: true });
    shared.promise.then(
      (result) => {
        release();
        resolve(result);
      },
      (error) => {
        release();
        reject(error);
      },
    );
  });
}

/**
 * Fetch Article content and update state. Matching same-process requests share
 * one cancellable operation; each caller can still detach without cancelling
 * active peers. Cross-process/out-of-order results are guarded by the Article
 * version comparison in `persistArticleOutcome`.
 */
export function fetchArticle(article: Article, options: CompleteArticleFetchOptions = {}): Promise<ArticleFetchResult> {
  validateArticleTimeout(options.timeout);
  if (options.signal?.aborted) return Promise.reject(abortError(options.signal));
  if (hasRenderableContent(article.content)) {
    return Promise.resolve({ article, status: 'skipped', persisted: false, current: false });
  }

  const key = articleFetchKey(article, options);
  let shared = inFlightArticleFetches.get(key);
  // The last subscriber may have cancelled just before a new caller arrives.
  // Do not attach that caller to an already-aborted operation while its Promise
  // is still unwinding; start a fresh, independently cancellable request.
  if (shared?.controller.signal.aborted || shared?.settled) {
    if (inFlightArticleFetches.get(key) === shared) inFlightArticleFetches.delete(key);
    shared = undefined;
  }
  if (!shared) {
    const controller = new AbortController();
    const promise = Promise.resolve().then(() => fetchArticleOnce(article, { ...options, signal: controller.signal }));
    shared = { controller, promise, subscribers: 0, settled: false };
    inFlightArticleFetches.set(key, shared);
    const operation = shared;
    void promise.then(
      () => {
        operation.settled = true;
        if (inFlightArticleFetches.get(key) === operation) inFlightArticleFetches.delete(key);
      },
      () => {
        operation.settled = true;
        if (inFlightArticleFetches.get(key) === operation) inFlightArticleFetches.delete(key);
      },
    );
  }

  return joinSharedFetch(shared, options.signal);
}

const MIN_RENDERABILITY_CANDIDATE_WINDOW = DEFAULT_ARTICLE_LIMIT;
const MAX_RENDERABILITY_CANDIDATE_WINDOW = DEFAULT_ARTICLE_LIMIT * 4;
const ARTICLE_RENDERABILITY_CURSOR_KEY = 'article-renderability-scan';
const SQLITE_BIND_CHUNK_SIZE = 400;

function validateArticleIds(articleIds: number[] | undefined): void {
  if (!articleIds) return;
  for (const articleId of articleIds) {
    if (!Number.isSafeInteger(articleId) || articleId <= 0) {
      throw new RangeError('articleIds must contain positive safe integers');
    }
  }
}

function renderabilityCandidateWindow(limit: number): number {
  return Math.min(Math.max(limit * 4, MIN_RENDERABILITY_CANDIDATE_WINDOW), MAX_RENDERABILITY_CANDIDATE_WINDOW);
}

function compareArticleFetchPriority(left: Article, right: Article): number {
  const leftPublishedAt = left.publishedAt?.getTime();
  const rightPublishedAt = right.publishedAt?.getTime();
  if (leftPublishedAt === undefined && rightPublishedAt !== undefined) return 1;
  if (leftPublishedAt !== undefined && rightPublishedAt === undefined) return -1;
  if (leftPublishedAt !== undefined && rightPublishedAt !== undefined && leftPublishedAt !== rightPublishedAt) {
    return rightPublishedAt - leftPublishedAt;
  }
  return right.id - left.id;
}

/**
 * Atomically claim a bounded keyset window for legacy renderability checks.
 * The cursor is durable so separate CLI invocations continue where the prior
 * one stopped instead of letting the same short, valid markup monopolize every
 * scan forever. Wrapping keeps newly eligible lower IDs reachable.
 */
function claimRenderabilityCandidates(condition: SQL | undefined, limit: number): Article[] {
  const db = getDb();
  return db.transaction(
    (tx) => {
      const state = tx
        .select({ lastId: maintenanceCursors.lastId })
        .from(maintenanceCursors)
        .where(eq(maintenanceCursors.key, ARTICLE_RENDERABILITY_CURSOR_KEY))
        .get();
      const cursor = Math.max(0, state?.lastId ?? 0);
      const afterCursor = tx
        .select()
        .from(articles)
        .where(and(condition, gt(articles.id, cursor)))
        .orderBy(asc(articles.id))
        .limit(limit)
        .all();
      const remaining = limit - afterCursor.length;
      const wrapped =
        remaining > 0
          ? tx
              .select()
              .from(articles)
              .where(and(condition, lte(articles.id, cursor)))
              .orderBy(asc(articles.id))
              .limit(remaining)
              .all()
          : [];
      const claimed = [...afterCursor, ...wrapped];
      const lastId = claimed.at(-1)?.id ?? 0;
      const updatedAt = new Date();

      tx.insert(maintenanceCursors)
        .values({ key: ARTICLE_RENDERABILITY_CURSOR_KEY, lastId, updatedAt })
        .onConflictDoUpdate({
          target: maintenanceCursors.key,
          set: { lastId, updatedAt },
        })
        .run();
      return claimed;
    },
    { behavior: 'immediate' },
  );
}

/** Fetch content for multiple Articles with concurrent processing. */
export async function fetchArticlesContent(
  articleIds?: number[],
  options: CompleteArticleFetchOptions & { limit?: number } = {},
): Promise<ArticleFetchResult[]> {
  validateArticleIds(articleIds);
  validateArticleTimeout(options.timeout);
  validateArticleConcurrency(options.concurrency);
  options.signal?.throwIfAborted();
  const db = getDb();
  const concurrency = options.concurrency ?? DEFAULT_CONCURRENCY;
  const skipBroken = options.skipBroken !== false;

  const shouldApplyDefaultLimit = !articleIds || articleIds.length === 0;
  const queryLimit = options.limit ?? (shouldApplyDefaultLimit ? DEFAULT_ARTICLE_LIMIT : undefined);
  if (queryLimit !== undefined && (!Number.isSafeInteger(queryLimit) || queryLimit < 0)) {
    throw new RangeError('limit must be a non-negative safe integer');
  }

  if (articleIds && articleIds.length > 0) {
    const uniqueArticleIds = [...new Set(articleIds)];
    const articlesToFetch: Article[] = [];
    let remaining = queryLimit;
    for (let offset = 0; offset < uniqueArticleIds.length; offset += SQLITE_BIND_CHUNK_SIZE) {
      if (remaining === 0) break;
      const chunk = uniqueArticleIds.slice(offset, offset + SQLITE_BIND_CHUNK_SIZE);
      const explicitQuery = db.select().from(articles).where(inArray(articles.id, chunk));
      const rows = remaining === undefined ? await explicitQuery : await explicitQuery.limit(remaining);
      articlesToFetch.push(...rows);
      if (remaining !== undefined) remaining -= rows.length;
    }
    return asyncPool(articlesToFetch, concurrency, (article) => fetchArticle(article, { ...options, skipBroken }));
  }

  const effectiveLimit = queryLimit ?? DEFAULT_ARTICLE_LIMIT;
  if (effectiveLimit === 0) return [];

  const retryCutoff = new Date(Date.now() - ARTICLE_CONTENT_RETRY_DELAY_MS);
  const eligibleUnbrokenState = and(
    eq(articles.isBroken, false),
    or(isNull(articles.contentFetchedAt), lt(articles.contentFetchedAt, retryCutoff)),
  );
  const eligibleState = skipBroken ? eligibleUnbrokenState : or(eligibleUnbrokenState, eq(articles.isBroken, true));
  const lacksStoredText = sql<boolean>`coalesce(trim(${articles.content}), '') = ''`;
  const hasStoredText = sql<boolean>`coalesce(trim(${articles.content}), '') <> ''`;
  const boundedStoredContent = sql<boolean>`length(cast(${articles.content} as blob)) <= ${MAX_CONTENT_SIZE}`;
  // Keep the OR grouped when this fragment is nested inside Drizzle's AND;
  // otherwise SQLite precedence would let entity-only rows bypass the state
  // and size guards that make this legacy scan safe.
  const looksLikeMarkupOrEntity = sql<boolean>`(instr(${articles.content}, '<') > 0 or instr(${articles.content}, '&') > 0)`;
  const order = [sql`${articles.publishedAt} DESC NULLS LAST`, desc(articles.id)] as const;

  // SQL can cheaply identify null/blank content but cannot reproduce
  // Turndown's renderability semantics. Claim a capped, durable keyset window
  // of bounded markup/entity candidates, then merge non-renderable legacy rows
  // with ordinary empty rows. Current write paths already reject such content;
  // the cursor makes progress across processes without unbounded parsing.
  // Final fetch priority remains newest-first.
  const candidateWindow = renderabilityCandidateWindow(effectiveLimit);
  const emptyContentArticles = await db
    .select()
    .from(articles)
    .where(and(eligibleState, lacksStoredText))
    .orderBy(...order)
    .limit(effectiveLimit);
  const boundedContentCandidates = claimRenderabilityCandidates(
    and(eligibleState, hasStoredText, boundedStoredContent, looksLikeMarkupOrEntity),
    candidateWindow,
  );
  const nonRenderableArticles = boundedContentCandidates.filter((article) => !hasRenderableContent(article.content));
  const articlesToFetch = [...emptyContentArticles, ...nonRenderableArticles]
    .sort(compareArticleFetchPriority)
    .slice(0, effectiveLimit);

  return asyncPool(articlesToFetch, concurrency, (article) => fetchArticle(article, { ...options, skipBroken }));
}