// src/db/schema/metering.ts (Drizzle + Postgres/TimescaleDB)
import { pgTable, text, integer, timestamp, index, unique } from "drizzle-orm/pg-core";

export const usageEvents = pgTable(
  "usage_events",
  {
    id: integer("id").primaryKey().generatedAlwaysAsIdentity(),
    eventId: text("event_id").notNull(),
    customerId: text("customer_id").notNull(),
    endpoint: text("endpoint").notNull(),
    quantity: integer("quantity").notNull(),
    occurredAt: timestamp("occurred_at", { withTimezone: true }).notNull(),
    ingestedAt: timestamp("ingested_at", { withTimezone: true }).notNull().defaultNow(),
  },
  (t) => ({
    eventIdUnique: unique("event_id_unique").on(t.eventId),
    customerTimeIdx: index("customer_time_idx").on(t.customerId, t.occurredAt),
  }),
);

export const usageAggregates = pgTable(
  "usage_aggregates",
  {
    customerId: text("customer_id").notNull(),
    periodStart: timestamp("period_start", { withTimezone: true }).notNull(),
    endpoint: text("endpoint").notNull(),
    totalQuantity: integer("total_quantity").notNull(),
    lastAggregatedAt: timestamp("last_aggregated_at", { withTimezone: true }).notNull(),
  },
  (t) => ({
    customerPeriodIdx: index("customer_period_idx").on(t.customerId, t.periodStart),
  }),
);

export type UsageEvent = typeof usageEvents.$inferSelect;
export type UsageAggregate = typeof usageAggregates.$inferSelect;

-- Migration: convert usage_events to a TimescaleDB hypertable
SELECT create_hypertable(
  'usage_events', 'occurred_at',
  chunk_time_interval => INTERVAL '1 day'
);

-- Continuous aggregate for the daily rollup (refreshed hourly)
CREATE MATERIALIZED VIEW usage_daily
WITH (timescaledb.continuous) AS
SELECT
  customer_id,
  endpoint,
  time_bucket('1 day', occurred_at) AS day,
  SUM(quantity) AS total_quantity
FROM usage_events
GROUP BY customer_id, endpoint, day;

// src/server/metering/ingest.ts
import { db } from "@/db";
import { usageEvents, usageAggregates } from "@/db/schema/metering";
import { and, eq, gte, sql } from "drizzle-orm";
import { stripe } from "@/lib/stripe";

export async function ingestUsageEvents(
  events: Array<{
    eventId: string;
    customerId: string;
    endpoint: string;
    quantity: number;
    occurredAt: Date;
  }>,
): Promise<{ inserted: number; deduped: number }> {
  // Idempotent batch insert via ON CONFLICT DO NOTHING
  const result = await db
    .insert(usageEvents)
    .values(
      events.map((e) => ({
        eventId: e.eventId,
        customerId: e.customerId,
        endpoint: e.endpoint,
        quantity: e.quantity,
        occurredAt: e.occurredAt,
      })),
    )
    .onConflictDoNothing({ target: usageEvents.eventId })
    .returning({ id: usageEvents.id });

  return {
    inserted: result.length,
    deduped: events.length - result.length,
  };
}

export async function reportDailyUsageToStripe(
  customerId: string,
  periodStart: Date,
): Promise<void> {
  const [{ total }] = await db
    .select({ total: sql<number>`COALESCE(SUM(total_quantity), 0)::int` })
    .from(usageAggregates)
    .where(
      and(
        eq(usageAggregates.customerId, customerId),
        gte(usageAggregates.periodStart, periodStart),
      ),
    );

  await stripe.subscriptionItems.createUsageRecord(
    customerId,
    { quantity: Number(total), timestamp: "now", action: "increment" },
  );
}

-- ClickHouse: usage_events MergeTree
CREATE TABLE usage_events (
  event_id        String,
  customer_id     String,
  endpoint        LowCardinality(String),
  quantity        UInt32,
  occurred_at     DateTime64(3, 'UTC'),
  ingested_at     DateTime DEFAULT now()
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(occurred_at)
ORDER BY (customer_id, occurred_at)
TTL occurred_at + INTERVAL 90 DAY;

-- AggregatingMergeTree for daily customer rollups
CREATE TABLE usage_daily
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(day)
ORDER BY (customer_id, endpoint, day)
AS SELECT
  customer_id,
  endpoint,
  toDate(occurred_at) AS day,
  sumState(quantity) AS total
FROM usage_events
GROUP BY customer_id, endpoint, day;

-- Q1: Customer usage for current billing period
SELECT
  endpoint,
  sumMerge(total) AS total_quantity
FROM usage_daily
WHERE customer_id = {customer_id:String}
  AND day >= {period_start:Date}
GROUP BY endpoint;

-- Q2: Plan-wide usage for yesterday
SELECT
  customer_id,
  sumMerge(total) AS total_quantity
FROM usage_daily
WHERE day = yesterday()
GROUP BY customer_id
ORDER BY total_quantity DESC
LIMIT 1000;

-- Q3: Per-endpoint deltas for the past 24 hours
SELECT
  endpoint,
  sumMerge(total) AS total_quantity,
  sumMerge(total) - lagInFrame(sumMerge(total)) OVER (
    PARTITION BY endpoint ORDER BY day
  ) AS delta
FROM usage_daily
WHERE day >= today() - 1
GROUP BY endpoint, day
ORDER BY endpoint, day DESC;