import { backfillAccounts, backfillDone, backfillOnce } from "../backfill";
import { all, batch, stmt } from "../db";
import { accountDb } from "../scoped-db";
import type { ScopedValue } from "../scoped-db/types";
import { scratchAccountIds } from "./directory";
import { periodOf } from "./record";

const INDEX = "usage-leaderboard";

export type UsageRow = { account_id: string; meter: string; quantity: number };
type Sql = { exec<T extends Record<string, ScopedValue>>(sql: string, ...params: ScopedValue[]): { toArray(): T[] } };

/** An account's counters and gauge peaks for a period, as its own AccountState stores them. */
export const PERIOD_USAGE_SQL = `SELECT account_id, meter, quantity FROM usage_periods WHERE period = ?1 AND quantity > 0
  UNION ALL SELECT account_id, meter, peak AS quantity FROM usage_gauges WHERE period = ?1 AND peak > 0`;

async function write(accountId: string, period: string, rows: UsageRow[], now: number): Promise<void> {
  const ids = [...new Set([accountId, ...rows.map((row) => row.account_id)])];
  await batch([
    stmt(
      `DELETE FROM usage_leaderboard WHERE period = ? AND account_id IN (${ids.map(() => "?").join(", ")})`,
      period,
      ...ids,
    ),
    ...rows.map((row) =>
      stmt(
        `INSERT INTO usage_leaderboard (period, meter, account_id, quantity, updated_at) VALUES (?, ?, ?, ?, ?)
         ON CONFLICT (period, meter, account_id) DO UPDATE SET quantity = excluded.quantity, updated_at = excluded.updated_at`,
        period,
        row.meter,
        row.account_id,
        row.quantity,
        now,
      ),
    ),
  ]);
}

/**
 * AccountState alarm, a few minutes after usage moves: replaces the account's rows for the current period, and
 * for the one before so usage just before a month ends still lands in that month's ranking.
 */
export async function publishUsageLeaderboard(accountId: string, sql: Sql, now: number): Promise<void> {
  const current = periodOf(now);
  const previous = periodOf(Date.parse(`${current}-01T00:00:00Z`) - 1);
  for (const period of [previous, current]) {
    await write(accountId, period, sql.exec<UsageRow>(PERIOD_USAGE_SQL, period).toArray(), now);
  }
}

async function copyAccount(accountId: string): Promise<void> {
  const now = Date.now();
  const period = periodOf(now);
  await write(accountId, period, await accountDb(accountId).all<UsageRow>(PERIOD_USAGE_SQL, period), now);
}

/** Cron step until done: copies this period's usage of accounts (and scratch ids) metered before the table. */
export async function backfillUsageLeaderboard(): Promise<void> {
  if (!(await backfillAccounts(INDEX, copyAccount))) return;
  await backfillOnce(`${INDEX}-scratch`, async () => {
    for (const accountId of await scratchAccountIds()) await copyAccount(accountId);
  });
}

export async function usageLeaderboardReady(): Promise<boolean> {
  return (await backfillDone(INDEX)) && (await backfillDone(`${INDEX}-scratch`));
}

/** The `perMeter` heaviest accounts on each meter in `period`. */
export function leaderboardRows(period: string, perMeter: number): Promise<UsageRow[]> {
  return all<UsageRow>(
    `SELECT account_id, meter, quantity FROM (
       SELECT account_id, meter, quantity, ROW_NUMBER() OVER (PARTITION BY meter ORDER BY quantity DESC) AS rank
       FROM usage_leaderboard WHERE period = ?
     ) WHERE rank <= ? ORDER BY meter, quantity DESC`,
    period,
    perMeter,
  );
}