backend/flexylead-route-work-items.mjs

1010        updated_at_iso: now.toISOString(),
1011      });
1012  }
1013  return rows.length;
1014}
1015
1016export function routingSnapshot(table, row, columns) {
1017  return {table,min_count:1,max_count:1,where:["id",...columns].map(col=>({col,op:row[col]==null?"is_null":"eq",...(row[col]==null?{}:{value:row[col]})}))};
1018}
1019
1020async function guardedMaintenanceUpdate(context, table, row, fields, dependencies = []) {
1021  return guardedRoutingWrites(context, {
1022    checks: [routingSnapshot(table,row,Object.keys(row).filter(k=>!["id","created_at","updated_at"].includes(k))), ...dependencies],
1023    mutations: [{operation:"update",table,id:Number(row.id),fields}],
1024  });
1025}
1026
1027export async function reconcileActiveAssignments(context, now, dryRun, limit = 300) {
1028  // Least-recently checked first: progress is durable across workers/restarts.
1029  // Every inspected row is touched, including rows whose state is unchanged.
1030  const assignments = await search(context, ASSIGNMENT_TABLE, [
1031    { col: "status", op: "in", value: [...ACTIVE_ASSIGNMENT_STATES] },
1032  ], "reconciled_at_iso asc", boundedInteger(limit, 300, 1, 300));
1033  let pending=[];
1034  const flush=async()=>{
1035    if(!pending.length)return;
1036    const batch=pending;pending=[];
1037    const committed=await guardedRoutingWrites(context, {checks:batch.flatMap(r=>r.checks),mutations:batch.flatMap(r=>r.mutations)});
1038    // One changing row must not starve the other rows in its batch.
1039    if(committed?.committed!==true)for(const item of batch)await guardedRoutingWrites(context, item);
1040  };
1041  const [calls, reservations, prospects] = await Promise.all([
1042    rowsById(
1043      context,
1044      "appels",
1045      assignments.filter((row) => clean(row.resource_type) === "missed_call").map((row) => row.resource_id),
1046      "id, handled_by, direction, status, ended_at, answered_at, talk_duration_seconds, followup_appel_id, followup_status, missed_pool_status, missed_claimed_by",
1047    ),
1048    rowsById(
1049      context,
1050      "reservations",
1051      assignments.filter((row) => ["ai_appointment", "planned_callback"].includes(clean(row.resource_type))).map((row) => row.resource_id),
1052      "id, status, commercial_outcome, outcome_at, claimed_by",
1053    ),
1054    rowsById(
1055      context,
1056      "prospects",
1057      assignments.filter((row) => clean(row.resource_type) === "assigned_lead").map((row) => row.resource_id),
1058      "id, statut, updated_at_iso",
1059    ),
1060  ]);
1061  for (const assignment of assignments) {
1062    const type = clean(assignment.resource_type);
1063    const rowId = Number(assignment.resource_id || 0);
1064    if (!rowId) {
1065      if(!dryRun)pending.push({checks:[routingSnapshot(ASSIGNMENT_TABLE,assignment,["status","resource_id"])],mutations:[{operation:"update",table:ASSIGNMENT_TABLE,id:Number(assignment.id),fields:{reconciled_at_iso:now.toISOString(),last_error:"invalid_resource_id"}}]});
1066      if(pending.length>=25)await flush();
1067      continue;
1068    }
1069    let status = "";
1070    let completedAt = null;
1071    let claimedBy = clean(assignment.claimed_by) || null;
1072    if (type === "missed_call") {
1073      const call = calls.get(clean(rowId));
1074      if (!call) status = "cancelled";
1075      else if (clean(call.followup_appel_id) || clean(call.followup_status) === "callback_started") status = "in_progress";
1076      else if (!isActionableMissedCall(call) || ["done", "discarded"].includes(clean(call.missed_pool_status).toLowerCase())) {
1077        status = "completed";
1078        completedAt = clean(call.ended_at) || now.toISOString();
1079      } else if (clean(call.missed_claimed_by)) {
1080        status = "claimed";
1081        claimedBy = clean(call.missed_claimed_by);
1082      } else status = "pooled";
1083    } else if (type === "ai_appointment" || type === "planned_callback") {
1084      const reservation = reservations.get(clean(rowId));
1085      if (!reservation) status = "cancelled";
1086      else if (TERMINAL_RESERVATION_STATES.has(clean(reservation.status).toLowerCase()) || TERMINAL_OUTCOMES.has(clean(reservation.commercial_outcome).toLowerCase())) {
1087        status = "completed";
1088        completedAt = clean(reservation.outcome_at) || now.toISOString();
1089      } else if (clean(reservation.status).toLowerCase() === "in_progress") status = "in_progress";
1090      else if (clean(reservation.claimed_by)) {
1091        status = "claimed";
1092        claimedBy = clean(reservation.claimed_by);
1093      }
1094    } else if (type === "assigned_lead") {
1095      const prospect = prospects.get(clean(rowId));
1096      const prospectStatus = clean(prospect?.statut).toLowerCase();