import { Injectable } from '@nestjs/common';
import { and, desc, eq, sql } from 'drizzle-orm';
import { randomBytes } from 'crypto';
import { CreateWebhookSchema, UpdateWebhookSchema, CreateConnectorSchema, UpdateConnectorSchema, MessageQuerySchema, PaginationQueryDto } from '@wms/shared';
import { z } from 'zod';
import { assertPublicUrl } from '@/common/util/net';
import { DbService } from '@/common/db/db.service';
import { webhookSubscriptions, webhookDeliveries, integrationConnectors, integrationMessages } from '@/db/schema';
import { AuditService } from '@/common/audit/audit.service';
import { EncryptionService } from '@/common/crypto/encryption.service';
import { requireRequestContext } from '@/common/context/request-context';
import { NotFoundError } from '@/common/errors/domain-error';
import { paginate, offsetOf, resolveStorerScope, serialize, qcol } from '@/common/util';

const SECRET_KEYS = /(secret|password|token|apikey|api_key)$/i;

/** Webhook subscriptions, connectors and the integration message log. */
@Injectable()
export class WebhooksService {
  constructor(
    private readonly db: DbService,
    private readonly audit: AuditService,
    private readonly crypto: EncryptionService,
  ) {}

  // ---- webhooks ----
  list() {
    const scope = resolveStorerScope();
    return this.db.run((tx) =>
      tx
        .select({ id: webhookSubscriptions.id, storerId: webhookSubscriptions.storerId, name: webhookSubscriptions.name, url: webhookSubscriptions.url, events: webhookSubscriptions.events, active: webhookSubscriptions.active, createdAt: webhookSubscriptions.createdAt, pending: sql<number>`(select count(*) from ${webhookDeliveries} d where d.subscription_id = ${qcol(webhookSubscriptions.id)} and d.status = 'PENDING')`.mapWith(Number), dead: sql<number>`(select count(*) from ${webhookDeliveries} d where d.subscription_id = ${qcol(webhookSubscriptions.id)} and d.status = 'DEAD')`.mapWith(Number) })
        .from(webhookSubscriptions)
        .where(scope ? eq(webhookSubscriptions.storerId, scope) : undefined)
        .orderBy(webhookSubscriptions.name),
    );
  }

  /** The signing secret is returned exactly once (on creation). */
  create(dto: z.infer<typeof CreateWebhookSchema>) {
    const ctx = requireRequestContext();
    return this.db.run(async (tx) => {
      await assertPublicUrl(dto.url, 'Webhook URL');
      const secret = dto.secret ?? randomBytes(24).toString('base64url');
      const [row] = await tx.insert(webhookSubscriptions).values({ tenantId: ctx.tenantId!, storerId: resolveStorerScope(dto.storerId) ?? null, name: dto.name, url: dto.url, secretEncrypted: this.crypto.encrypt(secret), events: dto.events, active: dto.active }).returning();
      await this.audit.log({ action: 'webhook.create', entityType: 'webhook_subscription', entityId: row.id, after: { ...dto, secret: undefined } }, tx);
      return { id: row.id, name: row.name, url: row.url, events: row.events, active: row.active, secret };
    });
  }

  update(id: string, dto: z.infer<typeof UpdateWebhookSchema>) {
    return this.db.run(async (tx) => {
      const [before] = await tx.select().from(webhookSubscriptions).where(eq(webhookSubscriptions.id, id)).limit(1);
      if (!before) throw new NotFoundError('Webhook', id);
      if (dto.url) await assertPublicUrl(dto.url, 'Webhook URL');
      const { secret, storerId, ...rest } = dto;
      const [row] = await tx.update(webhookSubscriptions).set({ ...rest, storerId: storerId === undefined ? undefined : storerId, secretEncrypted: secret ? this.crypto.encrypt(secret) : undefined }).where(eq(webhookSubscriptions.id, id)).returning();
      await this.audit.log({ action: 'webhook.update', entityType: 'webhook_subscription', entityId: id, before: { ...before, secretEncrypted: undefined }, after: { ...dto, secret: undefined } }, tx);
      return { id: row.id, name: row.name, url: row.url, events: row.events, active: row.active };
    });
  }

  remove(id: string) {
    return this.db.run(async (tx) => {
      const [row] = await tx.delete(webhookSubscriptions).where(eq(webhookSubscriptions.id, id)).returning();
      if (!row) throw new NotFoundError('Webhook', id);
      await this.audit.log({ action: 'webhook.delete', entityType: 'webhook_subscription', entityId: id }, tx);
      return { ok: true };
    });
  }

  deliveries(subscriptionId: string, q: PaginationQueryDto) {
    return this.db.run(async (tx) => {
      const where = eq(webhookDeliveries.subscriptionId, subscriptionId);
      const [{ count }] = await tx.select({ count: sql<number>`count(*)`.mapWith(Number) }).from(webhookDeliveries).where(where);
      const rows = await tx.select().from(webhookDeliveries).where(where).orderBy(desc(webhookDeliveries.createdAt)).limit(q.pageSize).offset(offsetOf(q));
      return paginate(rows, count, q);
    });
  }

  redeliver(deliveryId: string) {
    return this.db.run(async (tx) => {
      const [row] = await tx.update(webhookDeliveries).set({ status: 'PENDING', nextAttemptAt: new Date(), attempts: 0 }).where(eq(webhookDeliveries.id, deliveryId)).returning();
      if (!row) throw new NotFoundError('Delivery', deliveryId);
      return row;
    });
  }

  // ---- connectors ----
  listConnectors() {
    return this.db.run((tx) => tx.select().from(integrationConnectors).orderBy(integrationConnectors.name).then((rows) => rows.map((r) => ({ ...r, config: this.maskConfig(r.config) }))));
  }

  createConnector(dto: z.infer<typeof CreateConnectorSchema>) {
    const ctx = requireRequestContext();
    return this.db.run(async (tx) => {
      if (typeof dto.config.outboundUrl === 'string') await assertPublicUrl(dto.config.outboundUrl, 'Connector outbound URL');
      const [row] = await tx.insert(integrationConnectors).values({ tenantId: ctx.tenantId!, storerId: dto.storerId ?? null, type: dto.type, name: dto.name, config: this.encryptConfig(dto.config), active: dto.active }).returning();
      await this.audit.log({ action: 'connector.create', entityType: 'integration_connector', entityId: row.id, after: { ...dto, config: this.maskConfig(dto.config) } }, tx);
      return { ...row, config: this.maskConfig(row.config) };
    });
  }

  updateConnector(id: string, dto: z.infer<typeof UpdateConnectorSchema>) {
    return this.db.run(async (tx) => {
      const [before] = await tx.select().from(integrationConnectors).where(eq(integrationConnectors.id, id)).limit(1);
      if (!before) throw new NotFoundError('Connector', id);
      if (dto.config && typeof dto.config.outboundUrl === 'string') await assertPublicUrl(dto.config.outboundUrl, 'Connector outbound URL');
      const config = dto.config ? this.encryptConfig({ ...this.decryptConfig(before.config), ...dto.config }) : undefined;
      const [row] = await tx.update(integrationConnectors).set({ ...dto, storerId: dto.storerId === undefined ? undefined : dto.storerId, config }).where(eq(integrationConnectors.id, id)).returning();
      await this.audit.log({ action: 'connector.update', entityType: 'integration_connector', entityId: id, after: { ...dto, config: dto.config ? this.maskConfig(dto.config) : undefined } }, tx);
      return { ...row, config: this.maskConfig(row.config) };
    });
  }

  /** Decrypted config for runtime use by adapters. */
  decryptConfig(config: Record<string, unknown>): Record<string, unknown> {
    const out: Record<string, unknown> = {};
    for (const [k, v] of Object.entries(config)) out[k] = SECRET_KEYS.test(k) && typeof v === 'string' && v.startsWith('v1.') ? this.crypto.decrypt(v) : v;
    return out;
  }

  private encryptConfig(config: Record<string, unknown>): Record<string, unknown> {
    const out: Record<string, unknown> = {};
    for (const [k, v] of Object.entries(config)) out[k] = SECRET_KEYS.test(k) && typeof v === 'string' && !v.startsWith('v1.') ? this.crypto.encrypt(v) : v;
    return out;
  }

  private maskConfig(config: Record<string, unknown>): Record<string, unknown> {
    const out: Record<string, unknown> = {};
    for (const [k, v] of Object.entries(config)) out[k] = SECRET_KEYS.test(k) ? '********' : v;
    return out;
  }

  // ---- message log ----
  messages(q: z.infer<typeof MessageQuerySchema> & PaginationQueryDto) {
    const scope = resolveStorerScope(q.storerId);
    return this.db.run(async (tx) => {
      const where = and(
        q.direction ? eq(integrationMessages.direction, q.direction) : undefined,
        q.status ? eq(integrationMessages.status, q.status as never) : undefined,
        q.messageType ? eq(integrationMessages.messageType, q.messageType) : undefined,
        scope ? eq(integrationMessages.storerId, scope) : undefined,
        q.from ? sql`${integrationMessages.createdAt} >= ${q.from}` : undefined,
        q.to ? sql`${integrationMessages.createdAt} < ${q.to}::date + 1` : undefined,
      );
      const [{ count }] = await tx.select({ count: sql<number>`count(*)`.mapWith(Number) }).from(integrationMessages).where(where);
      const rows = await tx.select().from(integrationMessages).where(where).orderBy(desc(integrationMessages.createdAt)).limit(q.pageSize).offset(offsetOf(q));
      return paginate(rows.map((r) => ({ ...r, rawPayload: r.rawPayload ? `${r.rawPayload.slice(0, 500)}${r.rawPayload.length > 500 ? '…' : ''}` : null })), count, q);
    });
  }

  message(id: string) {
    return this.db.run(async (tx) => {
      const [m] = await tx.select().from(integrationMessages).where(eq(integrationMessages.id, id)).limit(1);
      if (!m) throw new NotFoundError('Message', id);
      return serialize(m);
    });
  }
}