Observabilidad, Jobs y Operación Técnica MVP

CREATE SCHEMA IF NOT EXISTS ops;
COMMENT ON SCHEMA ops IS 'FARO Connect · operación técnica · jobs, errores, health, backups';

CREATE TABLE IF NOT EXISTS ops.jobs (
  job_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  job_code text NOT NULL UNIQUE,
  name text NOT NULL,
  description text NOT NULL,

  job_type text NOT NULL CHECK (
    job_type IN ('scheduled', 'event_driven', 'manual', 'batch', 'system')
  ),

  queue_name text NOT NULL DEFAULT 'default',

  default_priority integer NOT NULL DEFAULT 5,
  max_attempts integer NOT NULL DEFAULT 3,
  timeout_seconds integer NOT NULL DEFAULT 300,

  retry_strategy text NOT NULL DEFAULT 'exponential' CHECK (
    retry_strategy IN ('none', 'fixed', 'exponential')
  ),

  retry_delay_seconds integer NOT NULL DEFAULT 60,

  is_company_scoped boolean NOT NULL DEFAULT true,
  is_active boolean NOT NULL DEFAULT true,

  schedule_cron text NULL,
  timezone text NOT NULL DEFAULT 'America/Argentina/Mendoza',

  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_ops_jobs_active
ON ops.jobs(is_active, job_type);

CREATE TABLE IF NOT EXISTS ops.job_runs (
  job_run_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  job_id uuid NULL,
  job_code text NOT NULL,

  company_id uuid NULL,

  queue_name text NOT NULL DEFAULT 'default',

  status text NOT NULL DEFAULT 'queued' CHECK (
    status IN ('queued', 'running', 'success', 'failed',
              'retrying', 'cancelled', 'dead_letter')
  ),

  priority integer NOT NULL DEFAULT 5,
  attempt integer NOT NULL DEFAULT 0,
  max_attempts integer NOT NULL DEFAULT 3,

  dedupe_key text NULL,

  payload jsonb NOT NULL DEFAULT '{}'::jsonb,
  result jsonb NOT NULL DEFAULT '{}'::jsonb,

  error_code text NULL,
  error_message text NULL,
  error_stack text NULL,

  started_at timestamptz NULL,
  finished_at timestamptz NULL,
  duration_ms integer NULL,

  scheduled_at timestamptz NOT NULL DEFAULT now(),
  next_retry_at timestamptz NULL,

  locked_by text NULL,
  locked_at timestamptz NULL,

  request_id text NULL,
  created_by uuid NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_job_runs_status_schedule
ON ops.job_runs(status, scheduled_at);

CREATE INDEX IF NOT EXISTS idx_job_runs_company_time
ON ops.job_runs(company_id, created_at DESC);

CREATE INDEX IF NOT EXISTS idx_job_runs_code_time
ON ops.job_runs(job_code, created_at DESC);

CREATE UNIQUE INDEX IF NOT EXISTS uq_job_runs_active_dedupe
ON ops.job_runs(company_id, job_code, dedupe_key)
WHERE dedupe_key IS NOT NULL
  AND status IN ('queued', 'running', 'retrying');

CREATE TABLE IF NOT EXISTS ops.job_locks (
  job_lock_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  lock_key text NOT NULL UNIQUE,
  company_id uuid NULL,

  locked_by text NOT NULL,
  locked_at timestamptz NOT NULL DEFAULT now(),
  expires_at timestamptz NOT NULL,

  metadata jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX IF NOT EXISTS idx_job_locks_expires
ON ops.job_locks(expires_at);

CREATE TABLE IF NOT EXISTS ops.job_schedules (
  job_schedule_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  job_id uuid NOT NULL REFERENCES ops.jobs(job_id) ON DELETE CASCADE,
  company_id uuid NULL,

  cron_expression text NOT NULL,
  timezone text NOT NULL DEFAULT 'America/Argentina/Mendoza',

  is_active boolean NOT NULL DEFAULT true,

  last_triggered_at timestamptz NULL,
  next_trigger_at timestamptz NULL,

  payload_template jsonb NOT NULL DEFAULT '{}'::jsonb,

  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_job_schedules_next
ON ops.job_schedules(is_active, next_trigger_at);

CREATE TABLE IF NOT EXISTS ops.error_events (
  error_event_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  company_id uuid NULL,
  user_id uuid NULL,

  source text NOT NULL CHECK (
    source IN ('frontend', 'api', 'worker', 'scheduler',
                'database', 'integration', 'ai_gateway',
                'storage', 'email_provider')
  ),

  severity text NOT NULL CHECK (
    severity IN ('low', 'medium', 'high', 'critical')
  ),

  error_code text NOT NULL,
  error_message text NOT NULL,

  job_run_id uuid NULL,
  request_id text NULL,

  entity_type text NULL,
  entity_id text NULL,

  stack_trace text NULL,
  context jsonb NOT NULL DEFAULT '{}'::jsonb,

  is_resolved boolean NOT NULL DEFAULT false,
  resolved_by uuid NULL,
  resolved_at timestamptz NULL,
  resolution_note text NULL,

  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_error_events_company_time
ON ops.error_events(company_id, created_at DESC);

CREATE INDEX IF NOT EXISTS idx_error_events_source_severity
ON ops.error_events(source, severity, created_at DESC);

CREATE INDEX IF NOT EXISTS idx_error_events_unresolved
ON ops.error_events(severity, created_at DESC)
WHERE is_resolved = false;

CREATE TABLE IF NOT EXISTS ops.health_snapshots (
  health_snapshot_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  status text NOT NULL CHECK (
    status IN ('healthy', 'degraded', 'unhealthy')
  ),

  db_status text NOT NULL,
  redis_status text NULL,
  storage_status text NULL,
  email_status text NULL,
  ai_status text NULL,

  pending_jobs integer NOT NULL DEFAULT 0,
  failed_jobs_24h integer NOT NULL DEFAULT 0,
  dead_letter_jobs integer NOT NULL DEFAULT 0,

  avg_job_duration_ms numeric(12,2) NULL,
  p95_api_latency_ms numeric(12,2) NULL,
  error_rate_5m numeric(12,4) NULL,

  details jsonb NOT NULL DEFAULT '{}'::jsonb,

  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_health_snapshots_time
ON ops.health_snapshots(created_at DESC);

CREATE INDEX IF NOT EXISTS idx_health_snapshots_status
ON ops.health_snapshots(status, created_at DESC);

CREATE TABLE IF NOT EXISTS ops.dead_letter_queue (
  dlq_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  job_run_id uuid NOT NULL REFERENCES ops.job_runs(job_run_id) ON DELETE CASCADE,
  job_code text NOT NULL,
  company_id uuid NULL,

  payload jsonb NOT NULL DEFAULT '{}'::jsonb,

  error_code text NOT NULL,
  error_message text NOT NULL,
  last_stack text NULL,

  attempts integer NOT NULL,
  first_failed_at timestamptz NOT NULL,
  last_failed_at timestamptz NOT NULL DEFAULT now(),

  status text NOT NULL DEFAULT 'pending' CHECK (
    status IN ('pending', 'investigating', 'requeued',
              'resolved', 'cancelled')
  ),

  resolution_note text NULL,
  resolved_by uuid NULL,
  resolved_at timestamptz NULL,

  created_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_dlq_status_time
ON ops.dead_letter_queue(status, last_failed_at DESC);

CREATE INDEX IF NOT EXISTS idx_dlq_company
ON ops.dead_letter_queue(company_id, status);

CREATE TABLE IF NOT EXISTS ops.alerts (
  alert_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  alert_code text NOT NULL,
  company_id uuid NULL,

  severity text NOT NULL CHECK (
    severity IN ('low', 'medium', 'high', 'critical')
  ),

  title text NOT NULL,
  description text NOT NULL,

  source_table text NULL,
  source_id uuid NULL,

  dedupe_key text NULL,
  fingerprint text NULL,

  channels text[] NOT NULL DEFAULT ARRAY[]::text[],
  recipients text[] NOT NULL DEFAULT ARRAY[]::text[],

  status text NOT NULL DEFAULT 'open' CHECK (
    status IN ('open', 'acknowledged', 'silenced', 'resolved')
  ),

  triggered_at timestamptz NOT NULL DEFAULT now(),
  acknowledged_at timestamptz NULL,
  resolved_at timestamptz NULL,

  context jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX IF NOT EXISTS idx_alerts_status_time
ON ops.alerts(status, triggered_at DESC);

CREATE UNIQUE INDEX IF NOT EXISTS uq_alerts_open_dedupe
ON ops.alerts(alert_code, dedupe_key)
WHERE status IN ('open', 'acknowledged')
  AND dedupe_key IS NOT NULL;

export type JobPayload = {
  companyId?: string;
  triggeredBy?: string | null;
  requestId?: string;
  payload: Record<string, unknown>;
};

export type JobHandlerContext = {
  jobRunId: string;
  jobCode: string;
  companyId?: string | null;
  requestId: string;
  attempt: number;
};

export type JobHandler = (
  payload: JobPayload,
  ctx: JobHandlerContext
) => Promise<Record<string, unknown> | void>;

import { ingestPendingFiles } from "@/src/ingest/ingestPendingFiles";
import { validateDataQuality } from "@/src/quality/validateDataQuality";
import { normalizeStaging } from "@/src/etl/normalizeStaging";
import { computeKpis } from "@/src/kpi/computeKpis";
import { evaluateRules } from "@/src/engine/evaluateRules";
import { runWorkflowChecker } from "@/src/workflow/workflowChecker";
import { recalculateScoreJob } from "@/src/score/recalculateScoreJob";
import { dispatchPendingNotifications } from "@/src/notifications/dispatchPendingNotifications";
import { generateWeeklyReportsJob } from "@/src/reports/generateWeeklyReportsJob";
import { cleanAiCache } from "@/src/ai/cleanAiCache";
import { applyRetention } from "@/src/ops/applyRetention";
import { checkBackups } from "@/src/ops/checkBackups";
import { createHealthSnapshot } from "@/src/ops/createHealthSnapshot";

import type { JobHandler } from "./ops.types";

export const jobRegistry: Record<string, JobHandler> = {
  "JOB-ING-001": ingestPendingFiles,
  "JOB-DQ-001": validateDataQuality,
  "JOB-NORM-001": normalizeStaging,
  "JOB-KPI-001": computeKpis,
  "JOB-RULE-001": evaluateRules,
  "JOB-WF-001": runWorkflowChecker,
  "JOB-SCORE-001": recalculateScoreJob,
  "JOB-NOTIF-001": dispatchPendingNotifications,
  "JOB-REPORT-001": generateWeeklyReportsJob,
  "JOB-AI-001": cleanAiCache,
  "JOB-RET-001": applyRetention,
  "JOB-BACKUP-001": checkBackups,
  "JOB-HEALTH-001": createHealthSnapshot
};

import { jobRegistry } from "./jobRegistry";
import { handleJobFailure } from "./handleJobFailure";

export async function runJob(params: {
  client: any;
  jobRunId: string;
  workerId: string;
}) {
  const started = Date.now();

  const jobResult = await params.client.query(
    `
    UPDATE ops.job_runs
    SET
      status = 'running',
      locked_by = $2,
      locked_at = now(),
      started_at = now(),
      attempt = attempt + 1,
      updated_at = now()
    WHERE job_run_id = $1
      AND status IN ('queued','retrying')
    RETURNING *
    `,
    [params.jobRunId, params.workerId]
  );

  const job = jobResult.rows[0];
  if (!job) return null;

  const handler = jobRegistry[job.job_code];
  if (!handler) {
    throw new Error(`JOB_HANDLER_NOT_FOUND: ${job.job_code}`);
  }

  try {
    // Setea contexto de empresa para RLS
    if (job.company_id) {
      await params.client.query(
        `SELECT set_config('app.company_id', $1, true)`,
        [job.company_id]
      );
    }

    const result = await handler(
      {
        companyId: job.company_id,
        requestId: job.request_id,
        payload: job.payload
      },
      {
        jobRunId: job.job_run_id,
        jobCode: job.job_code,
        companyId: job.company_id,
        requestId: job.request_id,
        attempt: job.attempt
      }
    );

    await params.client.query(
      `
      UPDATE ops.job_runs
      SET
        status = 'success',
        result = $2::jsonb,
        finished_at = now(),
        duration_ms = $3,
        updated_at = now()
      WHERE job_run_id = $1
      `,
      [job.job_run_id, JSON.stringify(result ?? {}), Date.now() - started]
    );

    return result;
  } catch (error: any) {
    await handleJobFailure({
      client: params.client,
      job,
      error,
      durationMs: Date.now() - started
    });
    return null;
  }
}

export async function handleJobFailure(params: {
  client: any;
  job: any;
  error: any;
  durationMs: number;
}) {
  const nextAttempt = Number(params.job.attempt ?? 0);
  const maxAttempts = Number(params.job.max_attempts ?? 3);

  const errorCode = params.error?.code ?? "JOB_FAILED";
  const errorMessage = params.error?.message ?? "Unknown job error";
  const stack = params.error?.stack ?? null;

  const shouldRetry = nextAttempt < maxAttempts;

  await params.client.query(
    `
    UPDATE ops.job_runs
    SET
      status = $2,
      error_code = $3,
      error_message = $4,
      error_stack = $5,
      finished_at = CASE WHEN $2 = 'dead_letter' THEN now() ELSE finished_at END,
      next_retry_at = CASE
        WHEN $2 = 'retrying' THEN now() + ($6::int || ' seconds')::interval
        ELSE NULL
      END,
      duration_ms = $7,
      updated_at = now()
    WHERE job_run_id = $1
    `,
    [
      params.job.job_run_id,
      shouldRetry ? "retrying" : "dead_letter",
      errorCode,
      errorMessage,
      stack,
      calculateRetryDelaySeconds(nextAttempt),
      params.durationMs
    ]
  );

  await params.client.query(
    `
    INSERT INTO ops.error_events (
      company_id, source, severity, error_code, error_message,
      job_run_id, stack_trace, context
    )
    VALUES ($1, 'worker', $2, $3, $4, $5, $6, $7::jsonb)
    `,
    [
      params.job.company_id,
      shouldRetry ? "medium" : "high",
      errorCode,
      errorMessage,
      params.job.job_run_id,
      stack,
      JSON.stringify({
        job_code: params.job.job_code,
        attempt: nextAttempt,
        max_attempts: maxAttempts
      })
    ]
  );

  // Si fue DLQ, materializa fila explícita
  if (!shouldRetry) {
    await params.client.query(
      `
      INSERT INTO ops.dead_letter_queue (
        job_run_id, job_code, company_id, payload,
        error_code, error_message, last_stack,
        attempts, first_failed_at
      )
      VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7, $8, now())
      `,
      [
        params.job.job_run_id,
        params.job.job_code,
        params.job.company_id,
        JSON.stringify(params.job.payload ?? {}),
        errorCode,
        errorMessage,
        stack,
        nextAttempt
      ]
    );
  }
}

function calculateRetryDelaySeconds(attempt: number) {
  // Backoff: 60s, 5m, 15m, 1h
  const delays = [60, 300, 900, 3600];
  return delays[Math.min(attempt - 1, delays.length - 1)];
}

export async function scheduleJobRun(params: {
  client: any;
  jobCode: string;
  companyId?: string | null;
  payload?: Record<string, unknown>;
  dedupeKey?: string | null;
  scheduledAt?: string | null;
}) {
  const job = await params.client.query(
    `
    SELECT * FROM ops.jobs
    WHERE job_code = $1 AND is_active = true
    LIMIT 1
    `,
    [params.jobCode]
  );

  if (!job.rows[0]) {
    throw new Error(`JOB_NOT_FOUND: ${params.jobCode}`);
  }

  const j = job.rows[0];

  const result = await params.client.query(
    `
    INSERT INTO ops.job_runs (
      job_id, job_code, company_id, queue_name,
      status, priority, max_attempts, dedupe_key,
      payload, scheduled_at
    )
    VALUES (
      $1, $2, $3, $4,
      'queued',
      $5, $6, $7,
      $8::jsonb,
      COALESCE($9::timestamptz, now())
    )
    ON CONFLICT DO NOTHING
    RETURNING job_run_id
    `,
    [
      j.job_id, j.job_code, params.companyId ?? null, j.queue_name,
      j.default_priority, j.max_attempts, params.dedupeKey ?? null,
      JSON.stringify(params.payload ?? {}), params.scheduledAt ?? null
    ]
  );

  return result.rows[0]?.job_run_id ?? null;
}

{
  "error": {
    "code": "PERMISSION_DENIED",
    "category": "PERMISSION_ERROR",
    "message": "No tenés permiso para realizar esta acción.",
    "request_id": "req_a8f3c1",
    "severity": "medium"
  }
}

HTTP/1.1 200 OK
Content-Type: application/json

{
  "status": "alive",
  "service": "faro-api",
  "version": "1.0.0",
  "timestamp": "2026-05-31T12:00:00-03:00"
}

HTTP/1.1 200 OK
Content-Type: application/json

{
  "status": "ready",
  "checks": {
    "database": "healthy",
    "redis": "healthy"
  },
  "timestamp": "2026-05-31T12:00:00-03:00"
}

HTTP/1.1 200 OK
Content-Type: application/json

{
  "status": "degraded",
  "checks": {
    "database": "healthy",
    "redis": "healthy",
    "storage": "healthy",
    "email_provider": "degraded",
    "ai_provider": "healthy"
  },
  "jobs": {
    "pending": 12,
    "failed_24h": 2,
    "dead_letter": 1,
    "avg_duration_ms": 842,
    "queue_depth_max": 18
  },
  "metrics": {
    "p95_api_latency_ms": 245,
    "error_rate_5m": 0.012
  },
  "timestamp": "2026-05-31T12:00:00-03:00"
}

export type HealthInputs = {
  db: "healthy" | "degraded" | "unhealthy";
  redis: "healthy" | "degraded" | "unhealthy";
  storage: "healthy" | "degraded" | "unhealthy";
  failedJobs24h: number;
  deadLetterJobs: number;
  errorRate5m: number;
};

export function classifyHealthStatus(
  i: HealthInputs
): "healthy" | "degraded" | "unhealthy" {
  if (i.db === "unhealthy") return "unhealthy";
  if (i.errorRate5m > 0.10) return "unhealthy";
  if (i.deadLetterJobs >= 5) return "unhealthy";

  if (i.redis !== "healthy") return "degraded";
  if (i.storage !== "healthy") return "degraded";
  if (i.failedJobs24h > 20) return "degraded";
  if (i.deadLetterJobs >= 1) return "degraded";
  if (i.errorRate5m > 0.05) return "degraded";

  return "healthy";
}

{
  "timestamp": "2026-05-31T12:00:00-03:00",
  "level": "info",
  "service": "faro-api",
  "environment": "production",
  "request_id": "req_a8f3c1",
  "company_id": "d4e1...empresa-demo-cuyo",
  "user_id": "7b2a...",
  "module": "score",
  "event": "score_recalculated",
  "message": "FARO Score recalculated for Empresa Demo Cuyo S.A.",
  "duration_ms": 842,
  "metadata": { "score": 66, "delta": -8 }
}

# HELP faro_api_request_count Total API requests
# TYPE faro_api_request_count counter
faro_api_request_count{method="GET",route="/api/v1/tensions",status="200"} 12483

# HELP faro_api_latency_ms API latency histogram
# TYPE faro_api_latency_ms histogram
faro_api_latency_ms_bucket{le="50"} 8421
faro_api_latency_ms_bucket{le="100"} 10987
faro_api_latency_ms_bucket{le="250"} 12104

# HELP faro_job_runs_total Jobs ejecutados
# TYPE faro_job_runs_total counter
faro_job_runs_total{job_code="JOB-SCORE-001",status="success"} 1284
faro_job_runs_total{job_code="JOB-SCORE-001",status="dead_letter"} 2

# HELP faro_queue_depth Jobs pendientes en queue
# TYPE faro_queue_depth gauge
faro_queue_depth{queue_name="notifications"} 12
faro_queue_depth{queue_name="score"} 0

# HELP faro_ai_cost_usd_daily Costo IA diario por empresa
# TYPE faro_ai_cost_usd_daily gauge
faro_ai_cost_usd_daily{company="empresa-demo-cuyo"} 1.84

export function withRequestId(handler: RequestHandler) {
  return async (req: Request, res: Response, next: NextFunction) => {
    const requestId = req.headers["x-request-id"]
      ?? `req_${crypto.randomUUID().slice(0, 8)}`;

    res.setHeader("x-request-id", requestId);

    await asyncLocalStorage.run({ requestId }, async () => {
      await handler(req, res, next);
    });
  };
}

CREATE TABLE IF NOT EXISTS ops.backup_checks (
  backup_check_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  resource_type text NOT NULL CHECK (
    resource_type IN ('database', 'storage', 'config', 'reports', 'evidence')
  ),

  status text NOT NULL CHECK (
    status IN ('ok', 'warning', 'failed')
  ),

  last_backup_at timestamptz NULL,
  last_restore_test_at timestamptz NULL,
  backup_size_bytes bigint NULL,

  provider text NULL,
  details jsonb NOT NULL DEFAULT '{}'::jsonb,

  checked_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_backup_checks_time
ON ops.backup_checks(resource_type, checked_at DESC);

CREATE TABLE IF NOT EXISTS ops.incidents (
  incident_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  severity text NOT NULL CHECK (
    severity IN ('SEV-1', 'SEV-2', 'SEV-3', 'SEV-4')
  ),

  status text NOT NULL DEFAULT 'open' CHECK (
    status IN ('open', 'investigating', 'mitigated', 'resolved', 'closed')
  ),

  title text NOT NULL,
  description text NOT NULL,

  impact_summary text NULL,
  root_cause text NULL,
  resolution_summary text NULL,

  started_at timestamptz NOT NULL DEFAULT now(),
  detected_at timestamptz NOT NULL DEFAULT now(),
  mitigated_at timestamptz NULL,
  resolved_at timestamptz NULL,
  closed_at timestamptz NULL,

  owner_user_id uuid NULL,
  related_alert_ids uuid[] NOT NULL DEFAULT ARRAY[]::uuid[],

  metadata jsonb NOT NULL DEFAULT '{}'::jsonb,

  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS idx_incidents_status_time
ON ops.incidents(status, started_at DESC);

CREATE INDEX IF NOT EXISTS idx_incidents_severity_open
ON ops.incidents(severity)
WHERE status IN ('open', 'investigating', 'mitigated');

CREATE TABLE IF NOT EXISTS ops.feature_flags (
  feature_flag_id uuid PRIMARY KEY DEFAULT gen_random_uuid(),

  flag_code text NOT NULL,
  company_id uuid NULL,   -- null = global

  enabled boolean NOT NULL DEFAULT false,
  config jsonb NOT NULL DEFAULT '{}'::jsonb,

  rollout_percentage integer NULL CHECK (
    rollout_percentage IS NULL OR (rollout_percentage BETWEEN 0 AND 100)
  ),

  description text NULL,
  created_by uuid NULL,

  created_at timestamptz NOT NULL DEFAULT now(),
  updated_at timestamptz NOT NULL DEFAULT now(),

  UNIQUE(flag_code, company_id)
);

CREATE INDEX IF NOT EXISTS idx_feature_flags_lookup
ON ops.feature_flags(flag_code, company_id);

export async function isFeatureEnabled(params: {
  client: any;
  flagCode: string;
  companyId?: string | null;
  defaultValue?: boolean;
}): Promise<boolean> {
  const rows = await params.client.query(
    `
    SELECT enabled, rollout_percentage
    FROM ops.feature_flags
    WHERE flag_code = $1
      AND (company_id = $2 OR company_id IS NULL)
    ORDER BY company_id NULLS LAST
    LIMIT 1
    `,
    [params.flagCode, params.companyId ?? null]
  );

  if (!rows.rows[0]) return params.defaultValue ?? false;

  const row = rows.rows[0];

  if (!row.enabled) return false;

  // Si hay rollout parcial, hash determinista por company
  if (row.rollout_percentage != null && params.companyId) {
    const bucket = hashToBucket(params.companyId);
    return bucket < row.rollout_percentage;
  }

  return true;
}

function hashToBucket(companyId: string): number {
  let hash = 0;
  for (let i = 0; i < companyId.length; i++) {
    hash = ((hash << 5) - hash + companyId.charCodeAt(i)) | 0;
  }
  return Math.abs(hash) % 100;
}

-- KPI diario de Empresa Demo Cuyo S.A.
SELECT
  c.company_id,
  c.timezone,
  date_trunc('day', sale_at AT TIME ZONE c.timezone) AS period_day,
  SUM(net_amount) AS net_sales
FROM faro.sales s
JOIN faro.companies c ON c.company_id = s.company_id
WHERE c.company_id = '...empresa-demo-cuyo'
  AND sale_at AT TIME ZONE c.timezone >= now() AT TIME ZONE c.timezone - interval '30 days'
GROUP BY 1, 2, 3
ORDER BY period_day DESC;