import { and, eq, gte, isNull, sql } from 'drizzle-orm';
import type { Database } from '@rareindex/database';
import { users, notifications } from '@rareindex/database';
import { logger, newId } from '@rareindex/shared';
import { sendMail, digestEmail, type DigestSection } from '@rareindex/notify';

const log = logger.child({ job: 'account.digest' });
const usd = (v: number) => new Intl.NumberFormat('en-US', { style: 'currency', currency: 'USD', maximumFractionDigits: 0 }).format(v);
const pct = (v: number | null) => (v === null ? '' : `${v > 0 ? '+' : ''}${(v * 100).toFixed(1)}%`);

/**
 * Daily/weekly digest: portfolio value + change, biggest movers in collections & watchlist, deals,
 * unread notifications. Sent once per period per user (tracked with a 'digest' notification row).
 */
export async function runDigests(db: Database, now = new Date()): Promise<number> {
  const weekday = now.getUTCDay();
  const members = await db.select({ id: users.id, email: users.email, name: users.name, prefs: users.preferences }).from(users).where(and(isNull(users.deletedAt), sql`${users.emailVerifiedAt} is not null`));
  let sent = 0;
  for (const m of members) {
    const p = (m.prefs ?? {}) as { digest?: string; digestWeekday?: number };
    const mode = p.digest ?? 'weekly';
    if (mode === 'off') continue;
    if (mode === 'weekly' && Number(p.digestWeekday ?? 1) !== weekday) continue;
    const periodDays = mode === 'daily' ? 1 : 7;
    const since = new Date(now.getTime() - periodDays * 86_400_000);
    const already = await db.select({ id: notifications.id }).from(notifications).where(and(eq(notifications.userId, m.id), eq(notifications.kind, 'digest'), gte(notifications.createdAt, new Date(now.getTime() - (periodDays * 24 - 2) * 3600_000)))).limit(1);
    if (already[0]) continue;

    const hist = (await db.execute(sql`
      select s.date::text as date, sum(s.value_usd)::float as v from collection_snapshots s join collections c on c.id = s.collection_id
      where c.user_id = ${m.id} and s.date >= ${since.toISOString().slice(0, 10)} group by s.date order by s.date
    `)) as unknown as Array<{ date: string; v: number }>;
    const first = hist[0]?.v ?? null;
    const last = hist[hist.length - 1]?.v ?? null;
    const portfolio = last !== null && last > 0 ? { valueUsd: usd(last), change: first && first > 0 ? pct(last / first - 1) : '—' } : null;

    const movers = (await db.execute(sql`
      select a.title, a.slug, st.riv_usd, st.change_7d, st.change_1d from (
        select ci.asset_id from collection_items ci join collections c on c.id = ci.collection_id where c.user_id = ${m.id}
        union select wi.target_id from watchlist_items wi join watchlists w on w.id = wi.watchlist_id where w.user_id = ${m.id} and wi.target_type = 'asset'
      ) x join assets a on a.id = x.asset_id join asset_stats st on st.asset_id = a.id
      where ${periodDays === 1 ? sql`st.change_1d` : sql`st.change_7d`} is not null
      order by abs(${periodDays === 1 ? sql`st.change_1d` : sql`st.change_7d`}) desc limit 6
    `)) as unknown as Array<{ title: string; slug: string; riv_usd: number | null; change_7d: number | null; change_1d: number | null }>;
    const deals = (await db.execute(sql`
      select a.title, a.slug, l.price_usd, l.discount_to_riv from listings l join assets a on a.id = l.asset_id join asset_stats s on s.asset_id = a.id
      where l.availability = 'available' and l.discount_to_riv <= -0.15 and l.discount_to_riv >= -0.5 and not ('riv_review' = any(l.flags)) and s.riv_confidence >= 0.5 and s.riv_sample_size >= 5
        and a.category_slug in (
          select distinct a2.category_slug from collection_items ci join collections c on c.id = ci.collection_id join assets a2 on a2.id = ci.asset_id where c.user_id = ${m.id}
          union select wi.target_id from watchlist_items wi join watchlists w on w.id = wi.watchlist_id where w.user_id = ${m.id} and wi.target_type = 'category'
        )
      order by l.discount_to_riv asc limit 5
    `)) as unknown as Array<{ title: string; slug: string; price_usd: number; discount_to_riv: number }>;
    const unread = (await db.execute(sql`select count(*)::int as n from notifications where user_id = ${m.id} and read_at is null and kind <> 'digest'`)) as unknown as Array<{ n: number }>;

    const site = (process.env.NEXT_PUBLIC_SITE_URL ?? 'https://www.rareindex.io').replace(/\/$/, '');
    const sections: DigestSection[] = [
      { heading: periodDays === 1 ? 'Movers today' : 'Movers this week', rows: movers.map((r) => ({ label: r.title, value: r.riv_usd ? usd(Number(r.riv_usd)) : '—', delta: pct(periodDays === 1 ? r.change_1d : r.change_7d), href: `${site}/asset/${r.slug}` })) },
      { heading: 'Deal Radar', rows: deals.map((r) => ({ label: r.title, value: usd(Number(r.price_usd)), delta: pct(Number(r.discount_to_riv)), href: `${site}/asset/${r.slug}` })) },
      { heading: 'Inbox', rows: unread[0]?.n ? [{ label: 'Unread notifications', value: String(unread[0].n), href: `${site}/notifications` }] : [] },
    ];
    if (!portfolio && sections.every((s) => s.rows.length === 0)) continue; // nothing to say — no empty digests
    const mail = digestEmail({ name: m.name, period: mode === 'daily' ? 'daily' : 'weekly', sections, portfolio });
    const res = await sendMail({ to: m.email, ...mail, tags: [{ name: 'kind', value: 'digest' }] });
    await db.insert(notifications).values({ id: newId('event'), userId: m.id, kind: 'digest', title: mail.subject, body: portfolio ? `Collection value ${portfolio.valueUsd} (${portfolio.change}).` : 'Your digest is ready.', href: '/collections', emailedAt: res.ok ? now : null, readAt: now });
    if (res.ok) sent++;
    else log.warn({ userId: m.id, err: res.error }, 'digest e-mail failed');
  }
  log.info({ sent }, 'digests processed');
  return sent;
}