alibaba/push/queue.js

// IndexedDB-backed push queue for inquiry ingest payloads.
// Stores pending/sent/failed items, supports a "nextRetry" backoff gate so
// the worker only picks up items whose retry window has elapsed.

const DB_NAME = 'inquiry_push';
const STORE = 'queue';
const VERSION = 1;

function openDb() {
  return new Promise((resolve, reject) => {
    const req = indexedDB.open(DB_NAME, VERSION);
    req.onupgradeneeded = () => {
      const db = req.result;
      if (!db.objectStoreNames.contains(STORE)) {
        db.createObjectStore(STORE, { keyPath: 'id' });
      }
    };
    req.onsuccess = () => resolve(req.result);
    req.onerror = () => reject(req.error);
  });
}

function store(db, mode) {
  return db.transaction(STORE, mode).objectStore(STORE);
}

function p(r) {
  return new Promise((res, rej) => {
    r.onsuccess = () => res(r.result);
    r.onerror = () => rej(r.error);
  });
}

function uuid() {
  if (globalThis.crypto?.randomUUID) return globalThis.crypto.randomUUID();
  return String(Date.now()) + '-' + Math.random().toString(36).slice(2);
}

export async function addItem(item) {
  const db = await openDb();
  try {
    const id = uuid();
    const rec = {
      id,
      payload: item.payload,
      inquiryId: item.inquiryId,
      status: 'pending',
      retries: 0,
      nextRetry: Date.now(),
      createdAt: Date.now(),
    };
    await p(store(db, 'readwrite').add(rec));
    return id;
  } finally {
    db.close();
  }
}

export async function getPending(limit = 5) {
  const db = await openDb();
  try {
    const all = await p(store(db, 'readonly').getAll());
    const now = Date.now();
    return all
      .filter(r => r.status === 'pending' && r.nextRetry <= now)
      .sort((a, b) => a.nextRetry - b.nextRetry)
      .slice(0, limit);
  } finally {
    db.close();
  }
}

export async function markStatus(id, status, opts = {}) {
  const db = await openDb();
  try {
    const st = store(db, 'readwrite');
    const r = await p(st.get(id));
    if (!r) return;
    Object.assign(r, { status, ...opts });
    await p(st.put(r));
  } finally {
    db.close();
  }
}

export async function remove(id) {
  const db = await openDb();
  try {
    await p(store(db, 'readwrite').delete(id));
  } finally {
    db.close();
  }
}

export async function getStats() {
  const db = await openDb();
  try {
    const all = await p(store(db, 'readonly').getAll());
    return all.reduce((acc, r) => {
      acc[r.status] = (acc[r.status] || 0) + 1;
      return acc;
    }, {});
  } finally {
    db.close();
  }
}