package meta

import (
	"context"
	"fmt"
	"time"
)

// BillingOpsDelta is one flushed increment of the server's in-process
// drive9_tenant_file_ops_total counter. Pods append deltas periodically; the
// billing worker aggregates them per hour. traffic_path is also a billing
// dimension: db9 and s3 traffic bill as separate messages. Cumulative is the
// absolute value at flush time and Gen the process-wide series-incarnation
// counter (a reactivated tenant's series gets a fresh generation, so a
// counter reset after eviction cannot collide with pre-eviction rows);
// together with pod/series identity they form the idempotency key that makes
// retried flushes at-most-once.
type BillingOpsDelta struct {
	TenantID    string
	OrgID       string
	Direction   string
	TrafficPath string
	Ops         int64
	Cumulative  int64
	Gen         int64
}

// BillingOpsSum is the aggregated ops usage of one tenant/direction/
// traffic-path over a time range.
type BillingOpsSum struct {
	TenantID    string
	OrgID       string
	Direction   string
	TrafficPath string
	Ops         int64
}

// billingOpsDeltaSchemaStatement is registered in metaInitSchemaStatements so
// every process opening the meta store converges on the same schema. The
// billing worker's own tables (ledger/outbox/snapshot/checkpoint) arrive with
// the worker PR (#126).
// uk_billing_ops_delta_point makes retried flushes at-most-once: cumulative
// is the series' absolute counter value at snapshot time, strictly increasing
// per (pod, series incarnation, series) while the process lives; the
// incarnation counter is process-wide monotonic and assigned when a series is
// (re)created, so an evicted-then-reactivated tenant's restarted counter
// cannot collide with pre-eviction rows.
const billingOpsDeltaSchemaStatement = `CREATE TABLE IF NOT EXISTS billing_ops_delta (
	id           BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
	tenant_id    VARCHAR(64) NOT NULL,
	org_id       VARCHAR(64) NOT NULL DEFAULT '',
	gen          BIGINT NOT NULL DEFAULT 0,
	direction    VARCHAR(8) NOT NULL,
	traffic_path VARCHAR(8) NOT NULL DEFAULT '',
	ops          BIGINT NOT NULL,
	cumulative   BIGINT NOT NULL,
	pod_id       VARCHAR(160) NOT NULL DEFAULT '',
	flushed_at   DATETIME(3) NOT NULL,
	created_at   DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
	UNIQUE KEY uk_billing_ops_delta_point (pod_id, tenant_id, gen, org_id, direction, traffic_path, cumulative),
	INDEX idx_billing_ops_delta_flushed (flushed_at),
	INDEX idx_billing_ops_delta_tenant (tenant_id, flushed_at)
)`

// InsertBillingOpsDeltas appends one batch of flushed counter deltas in a
// single transaction. flushedAt is truncated to milliseconds to match the
// DATETIME(3) column.
func (s *Store) InsertBillingOpsDeltas(ctx context.Context, podID string, flushedAt time.Time, deltas []BillingOpsDelta) (err error) {
	start := time.Now()
	defer observeMeta(ctx, "insert_billing_ops_deltas", start, &err)
	if len(deltas) == 0 {
		return nil
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback() }()
	stmt, err := tx.PrepareContext(ctx, `INSERT IGNORE INTO billing_ops_delta
		(tenant_id, org_id, gen, direction, traffic_path, ops, cumulative, pod_id, flushed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`)
	if err != nil {
		return err
	}
	defer func() { _ = stmt.Close() }()
	for _, d := range deltas {
		if d.TenantID == "" || d.Ops <= 0 {
			return fmt.Errorf("invalid billing ops delta: tenant=%q ops=%d", d.TenantID, d.Ops)
		}
		if _, err = stmt.ExecContext(ctx, d.TenantID, d.OrgID, d.Gen, d.Direction, d.TrafficPath, d.Ops, d.Cumulative,
			podID, flushedAt.UTC().Truncate(time.Millisecond)); err != nil {
			return err
		}
	}
	return tx.Commit()
}

// SumBillingOpsDeltas aggregates flushed ops per tenant/org/direction/
// traffic_path over [from, to). traffic_path is a billing dimension: db9 and
// s3 traffic are billed as separate messages.
func (s *Store) SumBillingOpsDeltas(ctx context.Context, from, to time.Time) (out []BillingOpsSum, err error) {
	start := time.Now()
	defer observeMeta(ctx, "sum_billing_ops_deltas", start, &err)
	rows, err := s.db.QueryContext(ctx, `SELECT tenant_id, org_id, direction, traffic_path, SUM(ops)
		FROM billing_ops_delta
		WHERE flushed_at >= ? AND flushed_at < ?
		GROUP BY tenant_id, org_id, direction, traffic_path
		ORDER BY tenant_id, org_id, direction, traffic_path`,
		from.UTC().Truncate(time.Millisecond), to.UTC().Truncate(time.Millisecond))
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var sum BillingOpsSum
		if err = rows.Scan(&sum.TenantID, &sum.OrgID, &sum.Direction, &sum.TrafficPath, &sum.Ops); err != nil {
			return out, err
		}
		out = append(out, sum)
	}
	return out, rows.Err()
}

// DeleteBillingOpsDeltasBefore removes flushed deltas older than the given
// time and returns the number of deleted rows, keeping the append-only table
// bounded.
func (s *Store) DeleteBillingOpsDeltasBefore(ctx context.Context, before time.Time) (deleted int64, err error) {
	start := time.Now()
	defer observeMeta(ctx, "delete_billing_ops_deltas", start, &err)
	res, err := s.db.ExecContext(ctx, `DELETE FROM billing_ops_delta WHERE flushed_at < ?`,
		before.UTC().Truncate(time.Millisecond))
	if err != nil {
		return 0, err
	}
	return res.RowsAffected()
}