// ❌ Broken pattern 1: The Non-Atomic Deadlock Trap
export async function naiveLock1(resource: string) {
  // If the process crashes between SETNX and EXPIRE, the lock NEVER expires!
  // Every subsequent worker is deadlocked forever.
  const acquired = await redis.setnx(`lock:${resource}`, '1');
  if (acquired) {
    await redis.expire(`lock:${resource}`, 30);
  }
}

// ❌ Broken pattern 2: The Silent Lock-Stealing Release Trap
export async function naiveLock2(resource: string) {
  const key = `lock:${resource}`;
  // Correct atomic acquisition...
  await redis.set(key, '1', 'NX', 'PX', 10000);

  // Do work that unexpectedly takes 12 seconds (GC pause, slow DB query)...
  await executeLongRunningJob();

  // 💥 DISASTER: Lock expired at 10s. Worker B acquired the lock at 10.1s.
  // Worker A finishes at 12s and calls DEL, DELETING WORKER B'S LOCK!
  // Worker C now acquires the lock, resulting in concurrent execution between B and C!
  await redis.del(key);
}

// ✅ Production Single-Instance Distributed Lock with Watchdog & Atomic Lua Release
import Redis from 'ioredis';
import crypto from 'crypto';

export class RedisLock {
  private redis: Redis;
  private key: string;
  private token: string;
  private ttlMs: number;
  private watchdogTimer: NodeJS.Timeout | null = null;

  constructor(redisClient: Redis, resourceKey: string, ttlMs = 10000) {
    this.redis = redisClient;
    this.key = `lock:${resourceKey}`;
    this.token = crypto.randomUUID(); // Unique ownership token
    this.ttlMs = ttlMs;
  }

  // 1. Atomic acquisition with unique token and millisecond lease (PX)
  async acquire(): Promise<boolean> {
    const result = await this.redis.set(this.key, this.token, 'NX', 'PX', this.ttlMs);
    if (result === 'OK') {
      this.startWatchdog();
      return true;
    }
    return false;
  }

  // 2. Background lease renewal (Heartbeat Watchdog)
  private startWatchdog() {
    const renewalInterval = Math.floor(this.ttlMs / 3);
    this.watchdogTimer = setInterval(async () => {
      // Atomic Lua script to renew TTL ONLY if we still own the token
      const renewLua = `
        if redis.call("get", KEYS[1]) == ARGV[1] then
          return redis.call("pexpire", KEYS[1], ARGV[2])
        else
          return 0
        end
      `;
      const renewed = await this.redis.eval(renewLua, 1, this.key, this.token, this.ttlMs);
      if (!renewed) {
        this.stopWatchdog();
      }
    }, renewalInterval);
  }

  private stopWatchdog() {
    if (this.watchdogTimer) {
      clearInterval(this.watchdogTimer);
      this.watchdogTimer = null;
    }
  }

  // 3. Atomic Lua release to guarantee we only release our own lock
  async release(): Promise<boolean> {
    this.stopWatchdog();
    const releaseLua = `
      if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
      else
        return 0
      end
    `;
    const result = await this.redis.eval(releaseLua, 1, this.key, this.token);
    return result === 1;
  }
}

-- ✅ PostgreSQL Schema enforcing Monotonic Fencing Tokens
CREATE TABLE inventory_allocations (
  product_id      BIGINT PRIMARY KEY,
  reserved_units  INT NOT NULL,
  fencing_token   BIGINT NOT NULL  -- Monotonically increasing revision counter
);

-- Atomically reserve inventory ONLY if the worker's fencing token is strictly newer
UPDATE inventory_allocations
SET 
  reserved_units = reserved_units + 5,
  fencing_token = :worker_token
WHERE 
  product_id = :product_id 
  AND fencing_token < :worker_token;  -- Rejects stale zombie writes!

-- ✅ Atomic Sliding Window Counter Rate Limiter (Lua Script)
-- KEYS[1]: Current window key (e.g. rate:user_100:1700000000)
-- KEYS[2]: Previous window key (e.g. rate:user_100:1699999940)
-- ARGV[1]: Window size in seconds (e.g. 60)
-- ARGV[2]: Current timestamp in seconds
-- ARGV[3]: Max allowed requests per window (e.g. 100)

local current_key = KEYS[1]
local prev_key = KEYS[2]
local window_size = tonumber(ARGV[1])
local now = tonumber(ARGV[2])
local max_limit = tonumber(ARGV[3])

local current_count = tonumber(redis.call('get', current_key) or "0")
local prev_count = tonumber(redis.call('get', prev_key) or "0")

-- Calculate weight of previous window based on elapsed time within current window
local time_into_current_window = now % window_size
local prev_weight = (window_size - time_into_current_window) / window_size
local estimated_count = math.floor(prev_count * prev_weight + current_count)

if estimated_count < max_limit then
  redis.call('incr', current_key)
  if current_count == 0 then
    redis.call('expire', current_key, window_size * 2)
  end
  return 1 -- Allowed
else
  return 0 -- Rate Limited (HTTP 429)
end

// ✅ Production Two-Phase Idempotent Payment Processor
export async function processPaymentIdempotent(idempotencyKey: string, paymentFn: () => Promise<PaymentResult>) {
  const lock = new RedisLock(redisClient, `idempotency:lock:${idempotencyKey}`, 15000);
  const resultKey = `idempotency:result:${idempotencyKey}`;

  // Step 1: Check if this idempotency key was already completed in the past
  const existingResult = await redisClient.get(resultKey);
  if (existingResult) {
    return JSON.parse(existingResult);
  }

  // Step 2: Acquire distributed lock to prevent concurrent in-flight duplicates
  const acquired = await lock.acquire();
  if (!acquired) {
    throw new Error('Concurrent request in progress. Please retry shortly.');
  }

  try {
    // Double-check result key inside lock
    const cachedInside = await redisClient.get(resultKey);
    if (cachedInside) return JSON.parse(cachedInside);

    // Step 3: Execute the authoritative payment operation
    const result = await paymentFn();

    // Step 4: Persist result with 24-hour retention
    await redisClient.set(resultKey, JSON.stringify(result), 'EX', 86400);
    return result;
  } finally {
    await lock.release();
  }
}