楽天 RMS の在庫更新を、無駄なく安全に回す — キュー設計・エンドポイント選定・売り越し対策の全実装

function buildAuthHeader(serviceSecret: string, licenseKey: string): string {
  const credentials = Buffer.from(`${serviceSecret}:${licenseKey}`).toString("base64")
  return `ESA ${credentials}`
}
 
async function callRmsApi<T>(endpoint: string, params: Record<string, unknown>): Promise<T> {
  const res = await fetch(`https://api.rms.rakuten.co.jp/es/${endpoint}`, {
    method: "POST",
    headers: {
      Authorization: buildAuthHeader(SERVICE_SECRET, LICENSE_KEY),
      "Content-Type": "application/json; charset=utf-8",
    },
    body: JSON.stringify(params),
  })
  if (!res.ok) throw new Error(`RMS API error: ${res.status}`)
  return res.json() as Promise<T>
}

// 在庫変動イベントをキューに積む（重複排除は後段で行う）
async function enqueueInventoryUpdate(
  itemUrl: string,
  skuId: string,
  newStock: number
) {
  await sqs.sendMessage({
    QueueUrl: INVENTORY_QUEUE_URL,
    MessageBody: JSON.stringify({ itemUrl, skuId, newStock, enqueuedAt: Date.now() }),
  })
}
 
// SQS Worker: Lambda で定期起動（例: 1 分ごと）
export async function handler(event: SQSEvent) {
  for (const record of event.Records) {
    const { itemUrl, skuId, newStock } = JSON.parse(record.body)
    try {
      await updateRmsInventory(itemUrl, skuId, newStock)
    } catch (err) {
      // Lambda の自動リトライに任せるか、DLQ に送る
      throw err
    }
    // 呼び出し間隔を開ける
    await delay(200)
  }
}
 
function delay(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms))
}

interface InventoryState {
  skuId: string           // PK
  itemUrl: string
  targetStock: number     // キューに積まれた最新の在庫数
  lastUpdatedAt: number   // エポックミリ秒
}
 
// 在庫変動イベントを「最新状態」として DynamoDB に upsert
async function upsertInventoryState(
  skuId: string,
  itemUrl: string,
  newStock: number,
  enqueuedAt: number
) {
  await dynamodb.update({
    TableName: "inventory-states",
    Key: { skuId },
    // より新しいイベントのみ上書き（古いイベントで巻き戻さない）
    UpdateExpression:
      "SET targetStock = :stock, itemUrl = :url, lastUpdatedAt = :ts",
    ConditionExpression:
      "attribute_not_exists(lastUpdatedAt) OR lastUpdatedAt < :ts",
    ExpressionAttributeValues: {
      ":stock": newStock,
      ":url": itemUrl,
      ":ts": enqueuedAt,
    },
  }).catch((err) => {
    if (err.name === "ConditionalCheckFailedException") {
      // より新しい状態がすでに書き込まれている → 何もしない（正常）
      return
    }
    throw err
  })
}

async function updateRmsInventory(itemUrl: string, skuId: string) {
  // 処理時点での最新在庫数を取得
  const state = await dynamodb.get({
    TableName: "inventory-states",
    Key: { skuId },
  })
  if (!state.Item) return // すでに別処理で消費済み
 
  const { targetStock } = state.Item as InventoryState
 
  await callRmsApi("inventory/2.0/updateInventory", {
    updateInventoryRequest: {
      inventoryList: [
        {
          itemUrl,
          inventorySkuList: [
            {
              skuId,
              inventoryStatus: "NORMAL",
              inventoryCount: targetStock,
            },
          ],
        },
      ],
    },
  })
}

// SQS キューの設定（CDK / CloudFormation での定義イメージ）
const dlq = new sqs.Queue(this, "InventoryDLQ", {
  retentionPeriod: Duration.days(14), // 2 週間保持
})
 
const mainQueue = new sqs.Queue(this, "InventoryQueue", {
  visibilityTimeout: Duration.seconds(30),
  deadLetterQueue: {
    queue: dlq,
    maxReceiveCount: 3, // 3 回失敗で DLQ へ
  },
})

// 夜間バッチ: 在庫マスタと RMS の在庫を突き合わせ
async function nightlyInventoryReconcile() {
  const allSkus = await fetchAllSkusFromMaster() // 社内在庫マスタ
 
  for (const sku of allSkus) {
    const masterStock = sku.currentStock
    const rmsStock = await fetchRmsInventory(sku.itemUrl, sku.skuId)
 
    if (masterStock !== rmsStock) {
      console.warn(`在庫不一致: ${sku.skuId} master=${masterStock} rms=${rmsStock}`)
      // 在庫マスタを正として RMS を上書き補正
      await enqueueInventoryUpdate(sku.itemUrl, sku.skuId, masterStock)
    }
  }
}