import { randomUUID } from 'node:crypto';
import { Queue, Worker } from 'bullmq';
import { Redis } from 'ioredis';
import { createApplicationDatabase, type Database } from '@atlas/database';
import { workerConfig } from './config.js';
import { OutboxDispatcher, type OutboxJob } from './outbox-dispatcher.js';
import { OzonCredentialError, OzonRequestError, verifyOzonCredentials } from './ozon-client.js';
import { decryptStoreSecret } from './store-secret.crypto.js';

const connection = new Redis(workerConfig.redisUrl, { maxRetriesPerRequest: null });
const database = createApplicationDatabase(workerConfig.databaseUrl);
const queue = new Queue<OutboxJob>('atlas-dispatch', { connection });
const dispatcher = new OutboxDispatcher(database, queue, `worker_${randomUUID()}`);

const worker = new Worker<OutboxJob>(
    'atlas-dispatch',
    async (job) => {
        if (job.name === 'store_credential_verification.requested') {
            await verifyStoreCredentials(database, job.data);
        }
    },
    { connection, concurrency: 5 },
);

worker.on('failed', (job, error) => {
    if (!job || job.attemptsMade < (job.opts.attempts ?? 1)) return;
    void markUnavailable(database, job.data, error);
});

async function verifyStoreCredentials(database: Database, job: OutboxJob) {
    const source = await database.workspaceTransaction(job.workspaceId, async (tx) => {
        await tx.query(
            `UPDATE atlas.operations SET status = 'running', started_at = COALESCE(started_at, now()),
             attempt_count = attempt_count + 1 WHERE id = $1 AND status IN ('queued', 'running')`,
            [job.payload.operation_id],
        );
        const result = await tx.query<{
            platform_code: string; platform_client_id: string; secret_ciphertext: Buffer; secret_nonce: Buffer;
        }>(
            `SELECT p.code AS platform_code, s.platform_client_id, secret.secret_ciphertext, secret.secret_nonce
             FROM atlas.operations operation JOIN atlas.stores s ON s.id = operation.resource_id AND s.deleted_at IS NULL
             JOIN atlas.platforms p ON p.id = s.platform_id JOIN atlas.store_secrets secret ON secret.id = $2
             WHERE operation.id = $1 AND operation.kind = 'store_credential_verification'
               AND secret.store_id = s.id AND secret.kind = 'seller_api_key' AND secret.revoked_at IS NULL`,
            [job.payload.operation_id, job.payload.secret_id ?? ''],
        );
        return result.rows[0] ?? null;
    });
    if (!source) return cancelStaleOperation(database, job);
    const apiKey = decryptStoreSecret(source.secret_ciphertext, source.secret_nonce, workerConfig.storeSecretEncryptionKey);
    try {
        if (source.platform_code.toUpperCase() !== 'OZON') throw new OzonRequestError(422);
        await verifyOzonCredentials(source.platform_client_id, apiKey);
        await completeVerification(database, job, true);
    } catch (error) {
        if (error instanceof OzonCredentialError) {
            await completeVerification(database, job, false, error.status);
            return;
        }
        if (error instanceof OzonRequestError) {
            await markRequestRejected(database, job, error.status);
            return;
        }
        throw error;
    }
}

async function cancelStaleOperation(database: Database, job: OutboxJob) {
    await database.workspaceTransaction(job.workspaceId, (tx) => tx.query(
        `UPDATE atlas.operations SET status = 'cancelled', completed_at = now(),
         error_code = 'CREDENTIAL_REPLACED', error_message = '凭据已更新，旧验证任务已取消'
         WHERE id = $1 AND status IN ('queued', 'running')`,
        [job.payload.operation_id],
    ));
}

async function completeVerification(
    database: Database,
    job: OutboxJob,
    valid: boolean,
    rejectedStatus?: 401 | 403,
) {
    await database.workspaceTransaction(job.workspaceId, async (tx) => {
        const current = await tx.query<{ id: string }>(
            `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2
             WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id
               AND secret.revoked_at IS NULL FOR UPDATE`,
            [job.payload.store_id, job.payload.secret_id ?? ''],
        );
        if (!current.rows[0]) return;
        const state = valid ? 'valid' : 'invalid';
        await tx.query(
            `UPDATE atlas.store_secrets SET status = $1::atlas.credential_status,
             last_verified_at = now(), updated_at = now() WHERE id = $2`,
            [state, job.payload.secret_id],
        );
        await tx.query(
            valid
                ? `UPDATE atlas.stores SET status = 'active', credential_status = 'valid',
                   credential_checked_at = now(), last_error_code = NULL, last_error_message = NULL,
                   last_error_at = NULL, revision = revision + 1, updated_at = now() WHERE id = $1`
                : `UPDATE atlas.stores SET status = 'credential_invalid', credential_status = 'invalid',
                   credential_checked_at = now(), last_error_code = $2,
                   last_error_message = $3, last_error_at = now(),
                   revision = revision + 1, updated_at = now() WHERE id = $1`,
            valid ? [job.payload.store_id] : [
                job.payload.store_id,
                rejectedStatus === 403 ? 'OZON_CREDENTIAL_FORBIDDEN' : 'OZON_CREDENTIAL_UNAUTHORIZED',
                rejectedStatus === 403
                    ? 'Ozon 拒绝访问：请检查 Client ID 与 API Key 是否属于同一账户，及密钥是否具备读取权限'
                    : 'Ozon 未接受该 Client ID 或 API Key',
            ],
        );
        await tx.query(
            valid
                ? `UPDATE atlas.operations SET status = 'succeeded', completed_at = now(),
                   result = jsonb_build_object('credential_status', 'valid'), error_code = NULL, error_message = NULL
                   WHERE id = $1 AND status IN ('queued', 'running')`
                : `UPDATE atlas.operations SET status = 'failed', completed_at = now(),
                   error_code = $2, error_message = $3
                   WHERE id = $1 AND status IN ('queued', 'running')`,
            valid ? [job.payload.operation_id] : [
                job.payload.operation_id,
                rejectedStatus === 403 ? 'OZON_CREDENTIAL_FORBIDDEN' : 'OZON_CREDENTIAL_UNAUTHORIZED',
                rejectedStatus === 403
                    ? 'Ozon 拒绝访问：请检查 Client ID 与 API Key 是否属于同一账户，及密钥是否具备读取权限'
                    : 'Ozon 未接受该 Client ID 或 API Key',
            ],
        );
    });
}

async function markRequestRejected(database: Database, job: OutboxJob, status: number) {
    await database.workspaceTransaction(job.workspaceId, async (tx) => {
        const current = await tx.query<{ id: string }>(
            `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2
             WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id
               AND secret.revoked_at IS NULL FOR UPDATE`,
            [job.payload.store_id, job.payload.secret_id ?? ''],
        );
        if (!current.rows[0]) return;
        const message = `Ozon 拒绝了验证请求（HTTP ${status}）；这不代表 API Key 无效`;
        await tx.query(
            `UPDATE atlas.operations SET status = 'failed', completed_at = now(),
             error_code = 'OZON_VERIFICATION_REQUEST_REJECTED', error_message = $2,
             result = jsonb_build_object('ozon_http_status', $3)
             WHERE id = $1 AND status IN ('queued', 'running')`,
            [job.payload.operation_id, message, status],
        );
        await tx.query(
            `UPDATE atlas.stores SET status = 'sync_error', last_error_code = 'OZON_VERIFICATION_REQUEST_REJECTED',
             last_error_message = $2, last_error_at = now(), revision = revision + 1, updated_at = now()
             WHERE id = $1 AND status = 'verifying'`,
            [job.payload.store_id, message],
        );
    });
}

async function markUnavailable(database: Database, job: OutboxJob, _error: Error) {
    await database.workspaceTransaction(job.workspaceId, async (tx) => {
        const current = await tx.query<{ id: string }>(
            `SELECT s.id FROM atlas.stores s JOIN atlas.store_secrets secret ON secret.id = $2
             WHERE s.id = $1 AND s.deleted_at IS NULL AND secret.store_id = s.id
               AND secret.revoked_at IS NULL FOR UPDATE`,
            [job.payload.store_id, job.payload.secret_id ?? ''],
        );
        if (!current.rows[0]) return;
        await tx.query(
            `UPDATE atlas.operations SET status = 'failed', completed_at = now(),
             error_code = 'OZON_VERIFICATION_UNAVAILABLE', error_message = '暂时无法验证 Ozon 凭据，请稍后重试'
             WHERE id = $1 AND status IN ('queued', 'running')`,
            [job.payload.operation_id],
        );
        await tx.query(
            `UPDATE atlas.stores SET status = 'sync_error', last_error_code = 'OZON_VERIFICATION_UNAVAILABLE',
             last_error_message = '暂时无法验证 Ozon 凭据，请稍后重试', last_error_at = now(),
             revision = revision + 1, updated_at = now() WHERE id = $1 AND status = 'verifying'`,
            [job.payload.store_id],
        );
    });
}

let timer: ReturnType<typeof setInterval> | undefined;
async function bootstrap() {
    await queue.waitUntilReady();
    await worker.waitUntilReady();
    await dispatcher.dispatchBatch();
    timer = setInterval(() => void dispatcher.dispatchBatch().catch((error: unknown) => {
        console.error('outbox_dispatch_failed', error instanceof Error ? error.message : error);
    }), 2_000);
    console.info('Atlas Worker is ready; consuming transactional-outbox events.');
}
void bootstrap().catch((error: unknown) => {
    console.error('worker_startup_failed', error instanceof Error ? error.message : error);
    process.exitCode = 1;
});
async function shutdown() {
    if (timer) clearInterval(timer);
    await worker.close(); await queue.close(); await database.close(); await connection.quit();
}
process.once('SIGINT', () => void shutdown());
process.once('SIGTERM', () => void shutdown());