feat(W3): expiry sync / full reconciliation sweep (schedule */5) + design + tested compute

- Single Code-node sweep: desired = CP-active paths UNION active AccessGrants; expiry excluded -> role/update rewrite; optional auto-disable (env-gated, default off) - disable guard: only when had scopes before (never fresh empty-role customer) - loop guard: no NocoDB Customers.status flip (avoids W1 re-provision loop) - test_w3.js mock-verified both AUTO_DISABLE modes; gen_w3.js builds JSON from w3-compute.js
2026-08-23 01:58:58 +08:00
parent 543f4a7885
commit a907a85a57
6 changed files with 435 additions and 0 deletions
@@ -0,0 +1,62 @@
+# W3 — Expiry Sync / 全量对账（schedule: every 5 min）
+
+> 定时任务，不是 webhook。负责「到期移除 scope、恢复客户、至空则禁用」。
+> 与 W1/W2 正交：W1/W2 是事件驱动（NocoDB 变更 → 重算单个 customer）；
+> W3 是周期性全量对账（扫所有 customer → desired vs actual → 修 AList / 回写 NocoDB）。
+> 幂等规则 §28：get current → compare → apply diff，重跑不产生重复写。
+
+## 触发
+
+- **Schedule Trigger**（`n8n-nodes-base.scheduleTrigger`），cron `*/5 * * * *`。
+- 每次**起手先 login** AList 换新 token（JWT_EXPIRE=2h，不缓存）。
+
+## 核心逻辑（desired-state）
+
+对每个**已有 `alist_role_id`** 的 customer：
+
+1. 读该 customer 全部 CustomerProducts（`where=(customer.Id,eq,${cid})&limit=1000`）。
+2. desired paths = 仅取 **status=active 且未过期** 的 CP 的 Product→Resources `alist_path` 集合并。
+   - 过期判定：`status==='expired'` 或 `expires_at < today`，一律**排除**。
+   - 有效 CP 无 `expires_at` → 视为长期有效，保留。
+3. 读 AList role 当前 scopes（`role/get?id=`）。
+4. `desired paths` vs `current paths` → 差异即要改。
+   - **到期移除**：current 有、desired 无 → 从 scopes 移除（整体重写 role/update）。
+   - **新生效加入**：desired 有、current 无 → 加入。
+5. 有变 → `POST /api/admin/role/update` 整体重写 `permission_scopes`（幂等）。
+6. **disable 政策（开关 `ALIST_AUTO_DISABLE_ON_EXPIRE`，默认 `false`）**：
+   - 若 `desired 为空` 且 该 customer NocoDB `status==='active'` →
+     回写 `status='expired'`，且（开关开时）`user/update disabled=true`。
+   - 默认关 = 首轮**只观察**，不自动禁用任何客户，避免误伤。
+
+## 实现形态（与 W2 的差异，需知晓）
+
+- **W2**（事件驱动）：Code 节点算 → 独立 HTTP 节点 apply，单客户，逐步可见。
+- **W3**（后台对账）：**单个 Code 节点内完成 read + write**（`this.helpers.httpRequest`），
+  每客户 try/catch 收集错误进 summary，最终返回一条 summary + 可选通知。
+  理由：① 无多 item 跨节点表达式陷阱（reconcile 输出 N 条会让 `$('X').item.json` 报错）；
+  ② 定时无人值守，原子化比逐步可见更稳；③ 幂等，重复跑无副作用。
+  代价：写操作在代码里，调试靠 per-customer `errors[]`。已权衡，接受。
+
+## Webhook loop 防护（关键）
+
+- NocoDB **webhook 触发的是 W1(Customers) 和 W2(CustomerProducts)**。
+- W3 回写要避开：
+  - 改 **CustomerProducts.sync_status** → 会触发 W2 webhook；W2 自带 "skip if already synced"
+    短路，安全。但 W3 应在**确无 diff** 时完全不写，减少空触发。
+  - 改 **Customers.status→expired** → 会触发 W1 webhook → 可能重新 provision → loop。
+    **规避**：W3 只在自己做了「去 scope 至空」时才翻 status；并且只设一个方向
+    （active→expired），同一同步周期内不回头。首期**开关默认关**，连 status 翻转也延后。
+  - 铁律（照 PROJECT_STATE）：不盲写 `last_sync` 时间戳；语义没变就不写。
+
+## 待确认决策点
+
+1. disable 政策是否首期启用？（建议否，先观察 1–2 轮）
+2. 是否顺带物化 AccessGrants（W3 写 product/manual 源之外）？—— 挂在 W2 验证后那项确认。
+3. reconciliation 结果日志：**不动 schema**（AccessLogs 是用户行为审计），先只靠 n8n execution + summary。要不要新开 `ReconcileLogs` 表，验证后再说。
+
+## 里程碑 / 测试
+
+- M1：以 `dryRun=true` 跑一次（只读，不写 AList/NocoDB），输出「将改的 diff」清单。
+- M2：造一条 `expires_at=昨天` 的 CP，观察 W3 将其 scope 移除（开关仍关，禁用不生效）。
+- M3：开 `ALIST_AUTO_DISABLE_ON_EXPIRE=true`，造一键全到期 customer，观察 role scopes 清空 + user disabled + status=expired。
+- 全部用 n8n execution error + summary JSON 验收。
@@ -0,0 +1,47 @@
+// Builds n8n/W3-expiry-sync.json from w3-compute.js (single source of truth).
+// Run: node n8n/gen_w3.js
+const fs = require('fs');
+const path = require('path');
+const jsCode = fs.readFileSync(path.join(__dirname, 'w3-compute.js'), 'utf8');
+
+const workflow = {
+  name: 'W3 - Expiry Sync (AList Role Reconciliation)',
+  nodes: [
+    {
+      parameters: {
+        rule: {
+          interval: [
+            { field: 'minutes', minutesInterval: 5 }
+          ]
+        }
+      },
+      id: 'trigger-schedule',
+      name: 'Every 5 minutes',
+      type: 'n8n-nodes-base.scheduleTrigger',
+      typeVersion: 1.2,
+      position: [0, 0]
+    },
+    {
+      parameters: { jsCode },
+      id: 'reconcile',
+      name: 'Reconcile & Apply (Expiry Sweep)',
+      type: 'n8n-nodes-base.code',
+      typeVersion: 2,
+      position: [220, 0]
+    }
+  ],
+  connections: {
+    'Every 5 minutes': {
+      main: [
+        [
+          { node: 'Reconcile & Apply (Expiry Sweep)', type: 'main', index: 0 }
+        ]
+      ]
+    }
+  },
+  settings: { executionOrder: 'v1' },
+  pinData: {}
+};
+
+fs.writeFileSync(path.join(__dirname, 'W3-expiry-sync.json'), JSON.stringify(workflow, null, 2) + '\n');
+console.log('wrote W3-expiry-sync.json (' + jsCode.length + ' chars jsCode)');
@@ -0,0 +1,137 @@
+// Standalone correctness test for w3-compute.js logic.
+// Mocks this.helpers.httpRequest + $env, runs the exact n8n Code-node code,
+// and asserts desired-union / diff / disable decisions.
+// Run:  node n8n/test_w3.js   (from Windows bash: node n8n/test_w3.js)
+const fs = require('fs');
+const path = require('path');
+
+const src = fs.readFileSync(path.join(__dirname, 'w3-compute.js'), 'utf8');
+
+// ---- fixtures (mirror NocoDB records / AList API shapes) ----
+// Resources: Id -> alist_path
+const RES = { 11: '/Class/Forex', 12: '/Class/Python', 13: '/Manual/Ebook' };
+// Products: Id -> [resourceIds]
+const PROD_RES = { 101: [11, 12], 102: [13] };
+// Customers: Id -> {alist_role_id, alist_user_id, status, disabled}
+const CUST = {
+  1: { alist_role_id: 501, alist_user_id: 901, status: 'active', disabled: false }, // qualified; has expired CP
+  2: { alist_role_id: 502, alist_user_id: 902, status: 'active', disabled: false }, // all expired -> should clear+disable(if on)
+  3: { alist_role_id: 503, alist_user_id: 903, status: 'active', disabled: false }, // no scopes now
+  4: { alist_role_id: null, alist_user_id: 904, status: 'active', disabled: false }, // not provisioned -> skip
+};
+// CustomerProducts: {Id, customer:Id, product:Id, expires_at, status}
+const CPS = {
+  1: [{ cid: 1, product: 101, expires_at: '2026-12-31', status: 'active' },     // live -> /Class/Forex,/Class/Python
+       { cid: 1, product: 102, expires_at: '2020-01-01', status: 'active' }],    // expired -> dropped
+  2: [{ cid: 2, product: 101, expires_at: '2020-01-01', status: 'active' }],     // expired -> empty
+  3: [],
+};
+// AccessGrants: active manual grant for cust 1 on /Manual/Ebook
+const GRANTS = {
+  1: [{ resource: 13, status: 'active', expires_at: null }],
+};
+// Current AList role scopes
+const ROLE_SCOPES = {
+  501: ['/Class/Forex', '/Stale/Old'],       // has stale -> remove; needs /Class/Python,/Manual/Ebook
+  502: ['/Class/Forex'],                      // all expired -> clear
+  503: [],
+};
+
+// ---- mock httpRequest keyed by URL ----
+const calls = { roleUpdate: [], userUpdate: [] };
+function httpRequest({ method, url, headers, body }) {
+  if (url.endsWith('/api/auth/login')) return { data: { token: 'tk-<TEST>' } };
+  const g = (m) => new RegExp(m).exec(url);
+  let m;
+  // per-table list endpoints
+  if (url.includes('/tables/RES/records?limit=10000')) {
+    return { list: Object.keys(RES).map((Id) => ({ Id: Number(Id), alist_path: RES[Id] })) };
+  }
+  if (url.includes('/tables/GRANT/records?limit=10000')) {
+    const list = [];
+    for (const cid of Object.keys(GRANTS)) for (const gg of GRANTS[cid]) list.push({ ...gg, customer: { Id: Number(cid) } });
+    return { list };
+  }
+  if (url.includes('/tables/CUST/records?limit=10000')) {
+    return { list: Object.keys(CUST).map((Id) => ({ Id: Number(Id), username: 'u' + Id, ...CUST[Id] })) };
+  }
+  // customerproducts by customer: /tables/CP/records?where=(customer.Id,eq,N)&limit=1000
+  if ((m = g('/tables/CP/records\\?where=\\(customer.Id,eq,([0-9]+)\\)&limit=1000'))) {
+    const cid = Number(m[1]);
+    return { list: (CPS[cid] || []).map((c) => ({ ...c, customer: { Id: cid } })) };
+  }
+  // product links: /tables/PROD/links/LNK/records/{pid}
+  if ((m = g('/tables/PROD/links/LNK/records/([0-9]+)'))) {
+    const pid = Number(m[1]);
+    return { list: (PROD_RES[pid] || []).map((Id) => ({ Id })) };
+  }
+  // role get
+  if ((m = g('/api/admin/role/get\\?id=([0-9]+)'))) {
+    const rid = Number(m[1]);
+    return { data: { id: rid, name: 'r' + rid, description: '', permission_scopes: (ROLE_SCOPES[rid] || []).map((path) => ({ path, permission: 0 })) } };
+  }
+  if (url.endsWith('/api/admin/role/update')) { calls.roleUpdate.push(body); return { data: null }; }
+  if (url.endsWith('/api/admin/user/update')) { calls.userUpdate.push(body); return { data: null }; }
+  throw new Error('no mock for ' + method + ' ' + url);
+}
+
+const $env = {
+  NOCODB_URL: 'http://nc', NOCODB_TOKEN: 't',
+  ALIST_URL: 'http://alist',
+  NOCODB_TABLE_CUSTOMERS: 'CUST', NOCODB_TABLE_CUSTOMERPRODUCTS: 'CP',
+  NOCODB_TABLE_PRODUCTS: 'PROD', NOCODB_TABLE_RESOURCES: 'RES',
+  NOCODB_TABLE_ACCESSGRANTS: 'GRANT', NOCODB_LINK_PROD_RESOURCES: 'LNK',
+  ALIST_ADMIN_USERNAME: 'a', ALIST_ADMIN_PASSWORD: 'p',
+  ALIST_AUTO_DISABLE_ON_EXPIRE: process.env.TEST_DISABLE === '1' ? 'true' : 'false',
+  W3_DRY_RUN: 'false',
+};
+// w3-compute uses plain (non-arrow) functions -> `this` resolves to globalThis
+// when called bare. Wrap in an async IIFE so the file's top-level `await` works
+// (n8n Code nodes also support top-level await). The file ends with `return {json:{...}}`
+// which becomes the IIFE's resolve value ( {json: payload} ).
+globalThis.helpers = { httpRequest };
+
+(async () => {
+  const runner = new Function('$env', '$json', 'return (async () => {\n' + src + '\n})();');
+  const json = await runner($env, {});
+  const out = json.json; // unwrap the {json: {...}} wrapper
+  console.log('SUMMARY:', JSON.stringify(out, null, 2));
+  console.log('SUMMARY:', JSON.stringify(out, null, 2));
+
+  // ---- assertions ----
+  const fail = (msg) => { console.error('FAIL: ' + msg); process.exitCode = 1; };
+  const ok = () => {};
+  const changed = out.changed;
+
+  const c1 = changed.find((c) => c.customerId === 1);
+  if (!c1) fail('cust1 should change'); else {
+    if (!c1.toRemove.includes('/Stale/Old')) fail('cust1: stale /Stale/Old not removed');
+    if (!c1.toAdd.includes('/Class/Python')) fail('cust1: /Class/Python not added');
+    if (!c1.toAdd.includes('/Manual/Ebook')) fail('cust1: manual /Manual/Ebook not added');
+    if (c1.toAdd.includes('/Class/Forex')) fail('cust1: /Class/Forex already present, should not be toAdd');
+    if (c1.applied.includes('role_scopes') === false) ok();
+  }
+  const c2 = changed.find((c) => c.customerId === 2);
+  if (!c2) fail('cust2 should change (all expired -> clear)'); else {
+    if (!c2.toRemove.includes('/Class/Forex')) fail('cust2: expired /Class/Forex not cleared');
+    if (c2.desiredPaths.length !== 0) fail('cust2 desiredPaths should be empty');
+    if (process.env.TEST_DISABLE === '1' && !c2.applied.includes('disable')) fail('cust2 should disable when auto-disable ON');
+    if (process.env.TEST_DISABLE !== '1' && c2.applied.includes('disable')) fail('cust2 must NOT disable when auto-disable OFF');
+  }
+  const c3 = changed.find((c) => c.customerId === 3);
+  if (c3) fail('cust3 (already empty scope) should NOT change');
+  if (changed.find((c) => c.customerId === 4)) fail('cust4 (no role) must be skipped');
+
+  // inspect actual writes
+  const ru = calls.roleUpdate.find((b) => b.id === 501);
+  if (ru) {
+    const paths = (ru.permission_scopes || []).map((s) => s.path);
+    if (!paths.includes('/Class/Forex')) fail('write cust1 lost /Class/Forex');
+    if (!paths.includes('/Manual/Ebook')) fail('write cust1 missing manual grant');
+    if (paths.includes('/Stale/Old')) fail('write cust1 kept stale');
+  } else fail('no role/update call for cust1');
+  if (process.env.TEST_DISABLE === '1' && !calls.userUpdate.find((b) => b.id === 902)) fail('cust2 user not disabled');
+
+  const nFail = Object.keys(calls.roleUpdate).length; // crude; we already assert specifics
+  console.log(process.exitCode ? '\n>>> TEST FAILED' : '\n>>> ALL CHECKS PASSED');
+})();
@@ -0,0 +1,124 @@
+// W3 - Expiry Sync / full reconciliation sweep (self-contained read+write).
+// Runs every 5 min. idempotent. get current -> compare -> apply diff.
+// Reads config via $env (works in n8n Code node).
+// dryRun: set W3_DRY_RUN=true env to only PRINT the plan (no AList/NocoDB writes).
+// AUTO_DISABLE: ALIST_AUTO_DISABLE_ON_EXPIRE=true -> disable AList user when desired empty.
+//   Default off = first rollout observes only. When on, ONLY AList user disabled (NocoDB Customers.status
+//   intentionally NOT flipped: Customers update webhook -> W1 could re-provision / re-enable = loop).
+
+const NC        = $env.NOCODB_URL;
+const NC_TOKEN  = $env.NOCODB_TOKEN;
+const ALIST     = $env.ALIST_URL;
+const T_CUST    = $env.NOCODB_TABLE_CUSTOMERS;
+const T_CUSTPROD= $env.NOCODB_TABLE_CUSTOMERPRODUCTS;
+const T_PROD    = $env.NOCODB_TABLE_PRODUCTS;
+const T_RES     = $env.NOCODB_TABLE_RESOURCES;
+const T_GRANT   = $env.NOCODB_TABLE_ACCESSGRANTS;
+const LNK_PROD_RES = $env.NOCODB_LINK_PROD_RESOURCES; // c4ko4x8kwi16swg
+
+const DRY_RUN   = ($env.W3_DRY_RUN || '').toLowerCase() === 'true';
+const AUTO_DISABLE = ($env.ALIST_AUTO_DISABLE_ON_EXPIRE || '').toLowerCase() === 'true';
+const today     = new Date().toISOString().slice(0, 10); // 'YYYY-MM-DD'
+
+async function ncGet(path) {
+  return await this.helpers.httpRequest({
+    method: 'GET', url: NC + path,
+    headers: { 'xc-token': NC_TOKEN }, json: true,
+  });
+}
+async function alistReq(method, path, body, token) {
+  return await this.helpers.httpRequest({
+    method, url: ALIST + path,
+    headers: token ? { Authorization: 'Bearer ' + token } : {},
+    body, json: true,
+  });
+}
+function linkId(v) {
+  if (v == null) return null;
+  if (typeof v === 'object' && !Array.isArray(v)) return v.Id ?? v.id ?? null;
+  if (Array.isArray(v)) return v[0]?.Id ?? v[0]?.id ?? null;
+  return v;
+}
+
+// login once
+const login = await alistReq('POST', '/api/auth/login', {
+  username: $env.ALIST_ADMIN_USERNAME,
+  password: $env.ALIST_ADMIN_PASSWORD,
+}, null);
+const token = login?.data?.token;
+if (!token) throw new Error('AList login failed');
+
+// index all Resources by Id -> alist_path
+const allRes = (await ncGet(`/api/v2/tables/${T_RES}/records?limit=10000`)).list || [];
+const resPath = {}; allRes.forEach(r => { resPath[r.Id] = r.alist_path; });
+
+// all access grants (materialized + manual), grouped by customer
+const grants = (await ncGet(`/api/v2/tables/${T_GRANT}/records?limit=10000`)).list || [];
+const gByCust = {};
+for (const g of grants) {
+  const cid = linkId(g.customer); if (!cid) continue;
+  (gByCust[cid] = gByCust[cid] || []).push(g);
+}
+
+const customers = (await ncGet(`/api/v2/tables/${T_CUST}/records?limit=10000`)).list || [];
+const changed = [], errors = [];
+
+for (const cust of customers) {
+  try {
+    if (!cust.alist_role_id) continue; // not provisioned
+    const cid = cust.Id;
+    // desired = CP-derived paths (active, not expired) union effective grants
+    const desired = {}; // path -> true
+    const cps = (await ncGet(`/api/v2/tables/${T_CUSTPROD}/records?where=(customer.Id,eq,${cid})&limit=1000`)).list || [];
+    for (const c of cps) {
+      if (c.status && c.status !== 'active') continue;
+      if (c.expires_at && c.expires_at < today) continue; // expired
+      const pid = linkId(c.product); if (!pid) continue;
+      const res = (await ncGet(`/api/v2/tables/${T_PROD}/links/${LNK_PROD_RES}/records/${pid}`)).list || [];
+      for (const r of res) { const p = resPath[r.Id]; if (p) desired[p] = true; }
+    }
+    for (const g of (gByCust[cid] || [])) {
+      if (g.status && g.status !== 'active') continue;
+      if (g.expires_at && g.expires_at < today) continue;
+      const p = resPath[linkId(g.resource)]; if (p) desired[p] = true;
+    }
+    const desiredPaths = Object.keys(desired).sort();
+
+    const role = await alistReq('GET', `/api/admin/role/get?id=${cust.alist_role_id}`, null, token);
+    const currentPaths = (role?.data?.permission_scopes || []).map(s => s.path);
+    const toRemove = currentPaths.filter(p => !desired[p]);
+    const toAdd    = desiredPaths.filter(p => !currentPaths.includes(p));
+    const scopesChanged = toAdd.length > 0 || toRemove.length > 0;
+    const nowEmpty = desiredPaths.length === 0;
+    // WF3: no remaining active grant + policy on + customer active + HAD scopes before -> disable.
+    // Guard currentPaths>0: never disable a fresh customer with an empty role (avoids
+    // killing a just-provisioned user before they buy anything; W2 won't re-enable them).
+    const shouldDisable = AUTO_DISABLE && nowEmpty && cust.status === 'active' && currentPaths.length > 0;
+
+    let applied = [];
+    if (scopesChanged) {
+      const newScopes = desiredPaths.map(p => ({ path: p, permission: 0 }));
+      if (!DRY_RUN) await alistReq('POST', '/api/admin/role/update',
+        { id: cust.alist_role_id, name: role?.data?.name, description: role?.data?.description,
+          permission_scopes: newScopes }, token);
+      applied.push('role_scopes');
+    }
+    if (shouldDisable) {
+      if (!DRY_RUN && cust.alist_user_id)
+        await alistReq('POST', '/api/admin/user/update', { id: cust.alist_user_id, disabled: true }, token);
+      applied.push('disable');
+    }
+    if (applied.length) {
+      changed.push({ customerId: cid, username: cust.username, roleId: cust.alist_role_id,
+        toAdd, toRemove, desiredPaths, applied });
+    }
+  } catch (e) {
+    errors.push({ customerId: cust.Id, username: cust.username, error: e.message });
+  }
+}
+
+return { json: {
+  dryRun: DRY_RUN, autoDisable: AUTO_DISABLE, at: new Date().toISOString(),
+  customersScanned: customers.length, customersChanged: changed.length,
+  changed, errors,
+} };