import { createHash } from 'node:crypto';
import { hexToBin, instantiateSecp256k1, type Secp256k1 } from '@bitauth/libauth';
import { toExtendedJson } from '@xo-cash/utils';
import { z } from 'zod';

import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts';

import type { BaseBroadcaster } from '../services/broadcaster.ts';
import { ApplicationError, UnauthorizedError } from '../errors/index.ts';
import type { Database } from '../services/storage/database.ts';
import type { RouteDefinition, RouteModule, RouteStream } from './types.ts';

/**
 * Schema to validate a single write resource.
 *
 * Each write targets one resource slot identified by (id, publicKey).
 * The publicKey owner signs the payload to prove write authority.
 */
const writeResource = z.object({
    id: z.string().min(1),
    publicKey: z.string().min(1),
    timestamp: z.number().positive(),
    signature: z.string().min(1),
    value: z.instanceof(Uint8Array),
});

/**
 * Schema to validate the write body.
 *
 * The body contains an array of write resources.
 */
const writeBody = z.object({
    resources: z.array(writeResource).min(1),
});

/**
 * Schema to validate and deduplicate resource IDs supplied in an application body.
 *
 * The body contains an array of resource IDs.
 */
const resourceIdsSchema = z.object({
    resourceId: z
        .array(z.string().min(1))
        .min(1, 'At least one resourceId is required')
        .transform((ids) => [ ...new Set(ids) ]),
});

type WriteResource = z.infer<typeof writeResource>;

/**
 * Resource data domain routes.
 *
 * A "resource" is a shared document identified by id. Multiple instances
 * (one per publicKey) can coexist within the same resource. Clients read,
 * write, subscribe to changes, and unsubscribe over the transport-neutral
 * RouteStream API.
 */
export class DataRoute implements RouteModule {
    /**
     * Promise to instantiate the secp256k1 library
     * This is a bit annoying, but keeping a single instance alive makes more sense than instantiating it for each verification.
     */
    private readonly secp256k1Promise: Promise<Secp256k1> = instantiateSecp256k1();

    constructor(
        private readonly database: Database,
        private readonly broadcaster: BaseBroadcaster,
        private readonly timestampWindowMs: number,
    ) {}

    /** Declare exact routes; each handler owns its stream behavior. */
    async getRoutes(): Promise<Array<RouteDefinition>> {
        return [
            {
                url: '/data/get',
                handler: this.getData.bind(this),
            },
            {
                url: '/data/write',
                handler: this.writeData.bind(this),
            },
            {
                url: '/data/subscribe',
                handler: this.subscribeData.bind(this),
            },
            {
                url: '/data/unsubscribe',
                handler: this.unsubscribeData.bind(this),
            },
        ];
    }

    /** Reads the requested resources from the database, returning all instances for each resource */
    async getData(stream: RouteStream): Promise<void> {
        const resourceIds = this.getResourceIds(stream);

        // Remove duplicates.
        const uniqueIds = [ ...new Set(resourceIds) ];

        // If there are no resource ids, return an empty array.
        if (uniqueIds.length === 0) {
            await stream.send([]);

            return;
        }

        // Read the data from the database.
        const rows = await this.database.db
            .selectFrom('resource_data')
            .select([ 'resource_id', 'public_key', 'blob', 'timestamp' ])
            .where('resource_id', 'in', uniqueIds)
            .orderBy('timestamp', 'asc')
            .execute();

        // Format the rows into the read resource responses.
        const formattedRows = rows.map((row) => ({
            resourceId: row.resource_id,
            publicKey: row.public_key,
            blob: new Uint8Array(row.blob),
            timestamp: row.timestamp,
        }));

        await stream.send(formattedRows);
    }

    /**
     * Authenticated batch write.
     *
     * Each resource in the batch is verified independently, then persisted and
     * broadcast to any subscribers listening on that resource's topic.
     */
    async writeData(stream: RouteStream): Promise<void> {
        const { resources } = writeBody.parse(stream.body);

        // Authenticate the whole batch before producing any storage side effects.
        // A single bad signature rejects the entire write — no partial commits.
        await Promise.all(resources.map((resource) => this.verifyWriteResource(resource)));

        const timestamp = Date.now();
        const rows = resources.map((resource) => ({
            resourceId: resource.id,
            publicKey: resource.publicKey,
            blob: Buffer.from(resource.value),
        }));

        // Upsert every row atomically; conflicts update blob and timestamp only.
        await this.database.db.transaction().execute(async (trx) => {
            for (const { resourceId, publicKey, blob } of rows) {
                await trx
                    .insertInto('resource_data')
                    .values({
                        resource_id: resourceId,
                        public_key: publicKey,
                        blob,
                        timestamp,
                    })
                    .onConflict((oc) =>
                        oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({
                            blob,
                            timestamp,
                        }))
                    .execute();
            }
        });

        // Format the rows into the written resource responses.
        const written = rows.map(({ resourceId, publicKey, blob }) => ({
            resourceId,
            instance: {
                publicKey,
                blob: new Uint8Array(blob),
                timestamp,
            },
        }));

        // Notify subscribers on each changed resource. Topic names are scoped per
        // resource id so clients only receive events for resources they joined.
        for (const { resourceId, instance } of written) {
            await this.broadcaster.publish(DataRoute.resourceTopic(resourceId), {
                type: 'instance-changed',
                data: { resourceId, ...instance },
            });
        }

        // Return the persisted instances so the writer can confirm what was stored.
        await stream.send({
            resources: written.map(({ resourceId, instance }) => ({
                id: resourceId,
                ...instance,
            })),
        });
    }

    /** Subscribe this connection to future changes for the requested resources. */
    async subscribeData(stream: RouteStream): Promise<void> {
        const resourceIds = this.getResourceIds(stream);

        const topics = resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId));

        // Subscribe registers the topics synchronously, then keeps this route
        // active until those topics are removed or the connection closes.
        await this.broadcaster.subscribe(stream, topics);
    }

    /**
     * Leave resource topics without closing the connection.
     *
     * Only available on bidirectional transports (WebSocket). SSE clients
     * unsubscribe implicitly by aborting the HTTP request.
     */
    async unsubscribeData(stream: RouteStream): Promise<void> {
        const resourceIds = this.getResourceIds(stream);

        // If the stream is not bidirectional, throw an error.
        if (!stream.bidirectional) {
            throw new ApplicationError(HTTP_STATUS_CODE_NOT_ACCEPTED, 'This route requires an existing bidirectional stream');
        }

        // Unsubscribe from the resource topics.
        await this.broadcaster.unsubscribe(
            stream,
            resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)),
        );
        await stream.send({});
    }

    /** Extract resource id list from the decoded request body. */
    private getResourceIds(stream: RouteStream): string[] {
        const body =
            typeof stream.body === 'object' && stream.body !== null && !Array.isArray(stream.body) ? (stream.body as Record<string, unknown>) : {};

        return resourceIdsSchema.parse(body).resourceId;
    }

    /**
     * Verify one write's timestamp freshness and secp256k1 signature.
     *
     * The client must sign the canonical payload built from timestamp, resource
     * id, and value — not the raw HTTP/WebSocket envelope.
     */
    private async verifyWriteResource(resource: WriteResource): Promise<void> {
        this.assertFreshTimestamp(resource.timestamp);

        if (!(await this.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) {
            throw new UnauthorizedError('Invalid resource signature');
        }
    }

    /**
     * Reject writes with stale timestamps to limit replay window.
     *
     * Both past and future timestamps outside the window are rejected.
     */
    private assertFreshTimestamp(timestamp: number): void {
        const age = Math.abs(Date.now() - timestamp);
        if (age > this.timestampWindowMs) {
            throw new UnauthorizedError('Timestamp outside allowed window');
        }
    }

    /**
     * Verify a secp256k1 signature.
     *
     * @param publicKeyHex - The public key to verify the signature against.
     * @param signatureHex - The signature to verify.
     * @param payload - The payload to verify the signature against.
     * @returns Whether the signature is valid.
     */
    private async verifySignature(publicKeyHex: string, signatureHex: string, payload: string): Promise<boolean> {
        const secp256k1 = await this.secp256k1Promise;

        try {
            // Convert the public key and signature to binary.
            const publicKey = hexToBin(publicKeyHex);
            const signature = hexToBin(signatureHex);

            // Create a SHA-256 hash of the payload.
            const messageHash = createHash('sha256').update(payload)
.digest();

            // Low-S normalization rejects malleable signature encodings.
            return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash);
        } catch {
            return false;
        }
    }

    /**
     * Build the transport-independent payload authenticated by each signature.
     *
     * canonicalBody ensures Uint8Array values hash consistently regardless of
     * whether the client sent them over HTTP or WebSocket.
     */
    private static canonicalWritePayload(resource: WriteResource): string {
        return `${resource.timestamp}${resource.id}${toExtendedJson(resource.value)}`;
    }

    /** Broadcaster topic for a single resource's instance-changed events. */
    private static resourceTopic(resourceId: string): string {
        return `resource:${resourceId}`;
    }
}