npm install @reaatech/media-pipeline-mcp-persistence

pnpm add @reaatech/media-pipeline-mcp-persistence

pnpm add ioredis

import { InMemoryPipelineStateStore } from '@reaatech/media-pipeline-mcp-persistence';
 
const store = new InMemoryPipelineStateStore();
 
// Create a pipeline run
await store.create({
  runId: 'run-42',
  pipelineId: 'product-photo',
  status: 'pending',
  tenantId: 'acme',
  pipelineDefHash: 'sha256-abc123',
  currentStepIndex: 0,
  steps: [
    {
      stepId: 'gen',
      operation: 'image.generate',
      inputs: { prompt: 'A product photo of a sneaker' },
      status: 'pending',
      attempts: 0,
      maxRetries: 2,
      artifactIds: [],
      costUsd: 0,
    },
  ],
  events: [],
  externalJobIds: {},
  version: 1,
  createdAt: new Date().toISOString(),
  updatedAt: new Date().toISOString(),
});
 
// Append events as the run progresses
await store.appendEvent('run-42', {
  kind: 'run-started',
  runId: 'run-42',
  at: Date.now(),
});
await store.appendEvent('run-42', {
  kind: 'step-started',
  runId: 'run-42',
  stepId: 'gen',
  at: Date.now(),
  attempt: 1,
});
await store.appendEvent('run-42', {
  kind: 'step-completed',
  runId: 'run-42',
  stepId: 'gen',
  at: Date.now(),
  artifactIds: ['artifact-99'],
  costUsd: 0.014,
});
await store.appendEvent('run-42', {
  kind: 'run-completed',
  runId: 'run-42',
  at: Date.now(),
  totalCostUsd: 0.014,
});
 
// Update with optimistic locking
await store.update(
  'run-42',
  { status: 'completed', completedAt: new Date().toISOString() },
  1, // expectedVersion — fails if another writer updated first
);
 
// Replay events (for reconnect/resume)
const events = await store.listEvents('run-42', /* sinceSeq */ 0);
 
// Query runs by tenant
const runs = await store.listRuns({
  tenantId: 'acme',
  status: 'completed',
  since: '2026-01-01T00:00:00Z',
  limit: 10,
  offset: 0,
});
 
// Cancel a run
await store.cancel('run-42', 'User requested cancellation');

import { Redis } from 'ioredis';
import { RedisPipelineStateStore } from '@reaatech/media-pipeline-mcp-persistence';
 
const redis = new Redis();
const store = new RedisPipelineStateStore({
  client: redis,
  prefix: 'mp',
  runTtlSeconds: 30 * 86400,   // 30 days
  lockTtlSeconds: 60,           // 60s lock TTL
  idempotencyTtlSeconds: 86400, // 24h
  externalJobTtlSeconds: 7 * 86400, // 7 days
  lockAcquireTimeoutMs: 5000,   // 5s max wait for lock
});
 
// Locked mutation
const result = await store.withLock('run-42', async (run) => {
  // Safe to mutate — exclusive lock held
  return performCriticalUpdate(run);
}, /* timeoutMs */ 10_000);

async function atomicStepComplete(runId: string, stepId: string, currentVersion: number) {
  await store.update(runId, {
    steps: updatedSteps,
    currentStepIndex: nextIndex,
  }, currentVersion); // Throws RunInProgressError if version mismatch
}

async function createIfAbsent(run: PipelineRun) {
  const existing = await store.listRuns({ idempotencyKey: run.idempotencyKey });
  if (existing.length > 0) return existing[0];
  await store.create(run);
  return run;
}

// When a provider webhook reports job completion
async function onJobComplete(provider: string, jobId: string, result: unknown) {
  const run = await store.findByExternalJobId(provider, jobId);
  if (!run) return; // Stale / unknown callback
  await store.appendEvent(run.runId, {
    kind: 'step-completed',
    runId: run.runId,
    stepId: currentStep(run).stepId,
    at: Date.now(),
    artifactIds: [result.id],
    costUsd: result.cost,
  });
}