API and Full Stack Coding Patterns Questions

import psycopg2
from contextlib import contextmanager
import redis
import json
r = redis.Redis()

@contextmanager
def db_txn(conn):
    cur = conn.cursor()
    try:
        yield cur
        conn.commit()
    except:
        conn.rollback()
        raise

def handle_webhook(conn, payload):
    event_id = payload['event_id']
    # Fast-path: try Redis lock to serialize concurrent handlers (optional)
    lock_key = f"lock:event:{event_id}"
    have_lock = r.set(lock_key, "1", nx=True, ex=30)
    if not have_lock:
        # another worker is processing; either wait or immediately return 200 to sender
        return 200

    try:
        with db_txn(conn) as cur:
            # Try to insert a row for this event. Unique constraint on event_id.
            try:
                cur.execute(
                  "INSERT INTO webhook_events(event_id, status, payload) VALUES (%s, %s, %s)",
                  (event_id, 'PROCESSING', json.dumps(payload))
                )
                # We "own" processing
                owned = True
            except psycopg2.IntegrityError:
                # row exists; check status
                conn.rollback()  # clear error state
                cur.execute("SELECT status FROM webhook_events WHERE event_id=%s FOR UPDATE", (event_id,))
                row = cur.fetchone()
                if row is None:
                    # shouldn't happen
                    return 500
                status = row[0]
                if status == 'PROCESSED':
                    return 200  # already done
                elif status == 'PROCESSING':
                    # concurrent processing ongoing; decide to wait, poll, or return success
                    return 200
                else:
                    # retryable state: claim by updating status (optimistic)
                    cur.execute("UPDATE webhook_events SET status=%s WHERE event_id=%s", ('PROCESSING', event_id))
                    owned = True

            # perform side effects (idempotent design recommended)
            perform_side_effects(cur, payload)

            # mark processed atomically
            cur.execute("UPDATE webhook_events SET status=%s WHERE event_id=%s", ('PROCESSED', event_id))

        return 200
    finally:
        if have_lock:
            r.delete(lock_key)

def perform_side_effects(cur, payload):
    # All DB writes happen using the same DB transaction/cursor so they are atomic with the status update.
    cur.execute("INSERT INTO orders(id, user_id, amount) VALUES (%s,%s,%s) ON CONFLICT DO NOTHING",
                (payload['order_id'], payload['user_id'], payload['amount']))
    # other writes...

class RequestLoader {
  constructor(batchFn) {
    this.queue = new Map(); // id -> [resolvers]
    this.scheduled = false;
    this.batchFn = batchFn;
  }
  load(id) {
    return new Promise(resolve => {
      if (!this.queue.has(id)) this.queue.set(id, []);
      this.queue.get(id).push(resolve);
      if (!this.scheduled) {
        this.scheduled = true;
        process.nextTick(() => this.flush());
      }
    });
  }
  async flush() {
    const ids = Array.from(this.queue.keys());
    const resolversMap = this.queue;
    this.queue = new Map();
    this.scheduled = false;
    const results = await this.batchFn(ids); // single DB query
    for (const id of ids) {
      const value = results[id] || null;
      for (const resolve of resolversMap.get(id)) resolve(value);
    }
  }
}

@RestController
@RequestMapping("/payments")
public class PaymentsController {
    private final IdempotencyRepo idemRepo;
    private final PaymentService paymentService;
    private final ObjectMapper mapper = new ObjectMapper();

    public PaymentsController(IdempotencyRepo idemRepo, PaymentService paymentService){
        this.idemRepo = idemRepo;
        this.paymentService = paymentService;
    }

    @PostMapping
    public ResponseEntity<?> createPayment(@RequestHeader("Idempotency-Key") String key,
                                           @RequestBody PaymentRequest req) {
        String clientId = req.getClientId(); // or from auth
        try {
            // Try to create a claim row atomically
            IdempotencyEntry entry = idemRepo.claim(key, clientId);
            if(entry.getStatus() == Status.COMPLETED){
                // Return stored response
                return ResponseEntity.status(entry.getHttpStatus())
                                     .body(mapper.readValue(entry.getResponseBody(), Object.class));
            }
            if(entry.getStatus() == Status.IN_PROGRESS){
                // Another request is processing. Respond with 409 or 202
                return ResponseEntity.status(HttpStatus.CONFLICT).body("Request in progress");
            }

            // Mark in-progress and process payment within a transaction
            entry.setStatus(Status.IN_PROGRESS);
            idemRepo.save(entry);

            PaymentResult result;
            try {
                result = paymentService.process(req); // external call; may throw
            } catch(Exception ex) {
                // mark failed so client can retry
                entry.setStatus(Status.FAILED);
                entry.setResponseBody(ex.getMessage());
                entry.setHttpStatus(HttpStatus.INTERNAL_SERVER_ERROR.value());
                idemRepo.save(entry);
                throw ex;
            }

            // on success, persist response for future duplicates
            String respBody = mapper.writeValueAsString(result);
            entry.setStatus(Status.COMPLETED);
            entry.setResponseBody(respBody);
            entry.setHttpStatus(HttpStatus.CREATED.value());
            idemRepo.save(entry);

            return ResponseEntity.status(HttpStatus.CREATED).body(result);

        } catch(IdempotencyConflictException cex){
            // insert-if-not-exists lost due to race; fetch existing
            IdempotencyEntry existing = idemRepo.findByKeyAndClient(key, clientId)
                .orElseThrow(() -> new ResponseStatusException(HttpStatus.INTERNAL_SERVER_ERROR));
            if(existing.getStatus() == Status.COMPLETED){
                return ResponseEntity.status(existing.getHttpStatus()).body(existing.getResponseBody());
            }
            return ResponseEntity.status(HttpStatus.CONFLICT).body("Request in progress");
        }
    }
}