modules/EcommerceFinance/Services/DingPayableWorkflowMirrorSynchronizer.php

@@ -5,6 +5,7 @@ namespace Modules\EcommerceFinance\Services;
 use Illuminate\Database\ConnectionInterface;
 use Illuminate\Database\Schema\Blueprint;
 use Illuminate\Support\Facades\Schema;
+use Illuminate\Support\Carbon;
 use Illuminate\Support\Str;
 use RuntimeException;
 use Throwable;
@@ -49,29 +50,28 @@ class DingPayableWorkflowMirrorSynchronizer
         }
 
         $mode = $full ? 'full' : 'incremental';
+        $stateCursor = $this->decodeJson($state['cursor'] ?? null);
+        $highWatermark = $this->cursorFromState($this->decodeJson($state['high_watermark'] ?? null));
         $resume = is_array($state)
             && ($state['run_mode'] ?? null) === $mode
             && in_array($state['status'] ?? null, ['running', 'failed'], true)
             && is_string($state['generation'] ?? null)
-            && is_array($state['cursor'] ?? null);
+            && $this->isCompatibleCursor($stateCursor);
         $generation = $resume ? (string) $state['generation'] : (string) Str::uuid();
-        $cursor = $resume
-            ? (string) ($state['cursor']['processInstanceId'] ?? '')
-            : ($full
-                ? ''
-                : (string) ($this->decodeJson($state['high_watermark'] ?? null)['processInstanceId'] ?? ''));
+        [$windowStart, $windowEnd] = $this->window($state, $mode, $resume);
+        $cursor = $resume ? $this->cursorFromState($stateCursor) : null;
         $rows = $resume ? (int) ($state['last_upserted_rows'] ?? 0) : 0;
         $batchSize = min(max($batchSize ?? (int) ($config['batch_size'] ?? 100), 1), 500);
         $maxBatches = max($maxBatches ?? (int) ($config['max_batches_per_run'] ?? 2), 0);
 
-        $this->start($dataset, $target, $source, $mode, $generation, $cursor, $resume);
+        $this->start($dataset, $target, $source, $mode, $generation, $resume, $windowStart, $windowEnd);
         $batches = 0;
 
         try {
             while (true) {
-                $workflows = $this->readWorkflows($source, $processCode, $cursor, $batchSize);
+                $workflows = $this->readWorkflows($source, $processCode, $cursor, $windowStart, $windowEnd, $batchSize);
                 if ($workflows === []) {
-                    $this->complete($dataset, $target, $mode, $generation, $rows, $cursor);
+                    $this->complete($dataset, $target, $mode, $generation, $rows, $cursor ?? $highWatermark);
 
                     return ['status' => 'succeeded', 'mode' => $mode, 'batches' => $batches, 'rows' => $rows, 'dataset' => $dataset];
                 }
@@ -116,7 +116,11 @@ class DingPayableWorkflowMirrorSynchronizer
                     ];
                 }
 
-                $cursor = (string) ($workflows[array_key_last($workflows)]['processInstanceId'] ?? $cursor);
+                $nextCursor = $this->cursorFromWorkflow($workflows[array_key_last($workflows)]);
+                if ($nextCursor === null) {
+                    throw new RuntimeException('ding_payable_workflow_missing_watermark');
+                }
+                $cursor = $nextCursor;
                 $rows += count($payloads);
                 $batches++;
                 $this->connection->transaction(function () use ($dataset, $target, $payloads, $cursor, $generation, $rows, $now): void {
@@ -130,7 +134,9 @@ class DingPayableWorkflowMirrorSynchronizer
                     $this->connection->table('business_mirror_table_states')
                         ->where('dataset_key', $dataset)
                         ->update([
-                            'cursor' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE),
+                            'cursor' => $cursor === null
+                                ? null
+                                : json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                             'last_upserted_rows' => $rows,
                             'updated_at' => $now,
                         ]);
@@ -184,7 +190,16 @@ class DingPayableWorkflowMirrorSynchronizer
         return $state;
     }
 
-    private function start(string $dataset, string $target, string $source, string $mode, string $generation, string $cursor, bool $resume): void
+    private function start(
+        string $dataset,
+        string $target,
+        string $source,
+        string $mode,
+        string $generation,
+        bool $resume,
+        ?Carbon $windowStart,
+        ?Carbon $windowEnd,
+    ): void
     {
         $now = now();
         $values = [
@@ -199,7 +214,12 @@ class DingPayableWorkflowMirrorSynchronizer
             'updated_at' => $now,
         ];
         if (! $resume) {
-            $values += ['cursor' => null, 'window_started_at' => null, 'window_ends_at' => null, 'last_upserted_rows' => 0];
+            $values += [
+                'cursor' => null,
+                'window_started_at' => $windowStart,
+                'window_ends_at' => $windowEnd,
+                'last_upserted_rows' => 0,
+            ];
         }
         $query = $this->connection->table('business_mirror_table_states')->where('dataset_key', $dataset);
         $query->exists()
@@ -207,7 +227,7 @@ class DingPayableWorkflowMirrorSynchronizer
             : $this->connection->table('business_mirror_table_states')->insert(['dataset_key' => $dataset, 'created_at' => $now] + $values);
     }
 
-    private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, string $cursor): void
+    private function complete(string $dataset, string $target, string $mode, string $generation, int $rows, ?array $cursor): void
     {
         $now = now();
         $this->connection->transaction(function () use ($dataset, $target, $mode, $generation, $rows, $cursor, $now): void {
@@ -219,7 +239,9 @@ class DingPayableWorkflowMirrorSynchronizer
             $values = [
                 'status' => 'succeeded',
                 'cursor' => null,
-                'high_watermark' => json_encode(['processInstanceId' => $cursor], JSON_UNESCAPED_UNICODE),
+                'high_watermark' => $cursor === null
+                    ? null
+                    : json_encode($cursor, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                 'last_succeeded_at' => $now,
                 'last_upserted_rows' => $rows,
                 'mirror_row_count' => $this->connection->table($target)->count(),
@@ -248,20 +270,97 @@ class DingPayableWorkflowMirrorSynchronizer
     }
 
     /** @return list<array<string,mixed>> */
-    private function readWorkflows(string $source, string $processCode, string $cursor, int $limit): array
+    private function readWorkflows(
+        string $source,
+        string $processCode,
+        ?array $cursor,
+        ?Carbon $windowStart,
+        ?Carbon $windowEnd,
+        int $limit,
+    ): array
     {
+        $watermark = 'COALESCE([updateAt], [createAt])';
         $where = '[processCode] = ?';
         $bindings = [$processCode];
-        if ($cursor !== '') {
-            $where .= ' AND [processInstanceId] > ?';
-            $bindings[] = $cursor;
+        if ($windowStart !== null && $windowEnd !== null) {
+            $where .= ' AND ' . $watermark . ' >= ? AND ' . $watermark . ' < ?';
+            $bindings[] = $this->sqlServerDateTime($windowStart);
+            $bindings[] = $this->sqlServerDateTime($windowEnd);
+        }
+        if ($cursor !== null) {
+            $where .= ' AND (' . $watermark . ' > ? OR (' . $watermark . ' = ? AND [processInstanceId] > ?))';
+            $bindings[] = (string) ($cursor['updateAt'] ?? '');
+            $bindings[] = (string) ($cursor['updateAt'] ?? '');
+            $bindings[] = (string) ($cursor['processInstanceId'] ?? '');
         }
         return $this->queries->forSource($source)->select(
-            'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId],[joinAt],[updateAt],[createAt],[processCode],[flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY [processInstanceId]',
+            'SELECT TOP (' . $limit . ') [finishTime],[businessId],[title],[result],[originatorUserId],[originatorDeptId],[originatorDeptName],[status],[bizAction],[processInstanceId], [joinAt], COALESCE([updateAt], [createAt]) AS [updateAt], [createAt], [processCode], [flowType] FROM [rpt].[ding_Workflow] WHERE ' . $where . ' ORDER BY ' . $watermark . ', [processInstanceId]',
             $bindings,
         );
     }
 
+    /** @return array{0:?Carbon,1:?Carbon} */
+    private function window(?array $state, string $mode, bool $resume): array
+    {
+        if ($mode === 'full') {
+            return [null, null];
+        }
+
+        if ($resume && ($state['window_started_at'] ?? null) !== null && ($state['window_ends_at'] ?? null) !== null) {
+            return [
+                Carbon::parse((string) $state['window_started_at']),
+                Carbon::parse((string) $state['window_ends_at']),
+            ];
+        }
+
+        $overlap = max((int) config('upstream_mirrors.table_sync.overlap_hours', 48), 1);
+        $lastSucceeded = ($state['last_succeeded_at'] ?? null) !== null
+            ? Carbon::parse((string) $state['last_succeeded_at'])
+            : now();
+
+        return [$lastSucceeded->copy()->subHours($overlap), now()->addSecond()];
+    }
+
+    /** @return array{updateAt:string,processInstanceId:string}|null */
+    private function cursorFromState(array $cursor): ?array
+    {
+        if (($cursor['updateAt'] ?? '') === '' && ($cursor['processInstanceId'] ?? '') === '') {
+            return null;
+        }
+
+        if (($cursor['updateAt'] ?? '') === '') {
+            return null;
+        }
+
+        return [
+            'updateAt' => (string) $cursor['updateAt'],
+            'processInstanceId' => (string) ($cursor['processInstanceId'] ?? ''),
+        ];
+    }
+
+    /** @param array<string,mixed> $workflow @return array{updateAt:string,processInstanceId:string}|null */
+    private function cursorFromWorkflow(array $workflow): ?array
+    {
+        $id = trim((string) ($workflow['processInstanceId'] ?? ''));
+        $updateAt = trim((string) ($workflow['updateAt'] ?? $workflow['createAt'] ?? ''));
+
+        if ($id === '' || $updateAt === '') {
+            return null;
+        }
+
+        return ['updateAt' => $updateAt, 'processInstanceId' => $id];
+    }
+
+    private function isCompatibleCursor(array $cursor): bool
+    {
+        return $cursor === [] || array_key_exists('updateAt', $cursor);
+    }
+
+    private function sqlServerDateTime(Carbon $value): string
+    {
+        return $value->format('Y-m-d\TH:i:s.v');
+    }
+
     /** @param list<string> $ids @param list<string> $columns @return list<array<string,mixed>> */
     private function readRelated(string $source, string $table, array $columns, array $ids, string $order): array
     {

tests/Unit/EcommerceFinance/DingPayableWorkflowMirrorSynchronizerTest.php

@@ -0,0 +1,199 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Services\DingPayableWorkflowMirrorSynchronizer;
+use Modules\EcommerceFinance\Services\UpstreamReadOnlyQuery;
+use Modules\EcommerceFinance\Services\UpstreamReadOnlyQueryFactory;
+use Tests\TestCase;
+
+class DingPayableWorkflowMirrorSynchronizerTest extends TestCase
+{
+    private FakeDingPayableWorkflowQuery $query;
+
+    private DingPayableWorkflowMirrorSynchronizer $synchronizer;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        config([
+            'database.connections.business_mirror' => [
+                'driver' => 'sqlite',
+                'database' => ':memory:',
+                'prefix' => '',
+            ],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.table_sync.overlap_hours' => 48,
+            'upstream_mirrors.ding_payable_workflows' => [
+                'dataset_key' => 'operations_replica.rpt.ding_payable_workflows',
+                'source_key' => 'operations_replica',
+                'process_code' => 'PROC-TEST',
+                'target_table' => 'ding_payable_workflow_mirror',
+                'batch_size' => 100,
+                'max_batches_per_run' => 2,
+            ],
+        ]);
+        DB::purge('business_mirror');
+
+        Schema::connection('business_mirror')->create('business_mirror_table_states', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->string('source_key');
+            $table->string('source_object');
+            $table->string('target_table');
+            $table->string('status')->default('pending');
+            $table->string('run_mode')->nullable();
+            $table->string('generation')->nullable();
+            $table->json('cursor')->nullable();
+            $table->json('high_watermark')->nullable();
+            $table->dateTime('window_started_at')->nullable();
+            $table->dateTime('window_ends_at')->nullable();
+            $table->dateTime('initial_completed_at')->nullable();
+            $table->dateTime('last_full_succeeded_at')->nullable();
+            $table->dateTime('last_attempted_at')->nullable();
+            $table->dateTime('last_succeeded_at')->nullable();
+            $table->unsignedBigInteger('last_upserted_rows')->default(0);
+            $table->unsignedBigInteger('mirror_row_count')->default(0);
+            $table->string('last_error_code')->nullable();
+            $table->timestamps();
+        });
+
+        $this->query = new FakeDingPayableWorkflowQuery();
+        $factory = new class($this->query) extends UpstreamReadOnlyQueryFactory {
+            public function __construct(private readonly UpstreamReadOnlyQuery $query) {}
+
+            public function forSource(string $source): UpstreamReadOnlyQuery
+            {
+                return $this->query;
+            }
+        };
+        $this->synchronizer = new DingPayableWorkflowMirrorSynchronizer($factory);
+    }
+
+    public function test_full_sync_resumes_with_update_time_and_process_id_tie_breaker(): void
+    {
+        $sameUpdateAt = '2026-08-29 10:00:00.1234567';
+        $this->query->workflowBatches = [[
+            $this->workflow('A-process', $sameUpdateAt),
+            $this->workflow('B-process', $sameUpdateAt),
+        ]];
+
+        $result = $this->synchronizer->sync(true, 2, 1);
+
+        $this->assertSame('running', $result['status']);
+        $this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $this->query->calls[0]['sql']);
+        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
+        $this->assertSame([
+            'updateAt' => $sameUpdateAt,
+            'processInstanceId' => 'B-process',
+        ], json_decode((string) $state->cursor, true));
+
+        $this->query->calls = [];
+        $this->query->workflowBatches = [[
+            $this->workflow('C-process', $sameUpdateAt),
+        ]];
+
+        $result = $this->synchronizer->sync(true, 2, 0);
+
+        $this->assertSame('succeeded', $result['status']);
+        $this->assertStringContainsString(
+            'COALESCE([updateAt], [createAt]) > ? OR (COALESCE([updateAt], [createAt]) = ? AND [processInstanceId] > ?',
+            $this->query->calls[0]['sql'],
+        );
+        $this->assertSame([$sameUpdateAt, $sameUpdateAt, 'B-process'], array_slice($this->query->calls[0]['bindings'], -3));
+        $this->assertSame(3, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count());
+    }
+
+    public function test_incremental_sync_uses_overlap_window_and_keeps_fixed_process_code(): void
+    {
+        $this->query->workflowBatches = [[
+            $this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')),
+        ]];
+        $this->synchronizer->sync(true, 10, 0);
+
+        $this->query->calls = [];
+        $this->query->workflowBatches = [[
+            $this->workflow('new-process', now()->format('Y-m-d H:i:s.u'), 'RUNNING'),
+        ]];
+
+        $result = $this->synchronizer->sync(false, 10, 0);
+
+        $this->assertSame('succeeded', $result['status']);
+        $sql = $this->query->calls[0]['sql'];
+        $this->assertStringContainsString('[processCode] = ?', $sql);
+        $this->assertStringContainsString('COALESCE([updateAt], [createAt]) >= ? AND COALESCE([updateAt], [createAt]) < ?', $sql);
+        $this->assertStringContainsString('ORDER BY COALESCE([updateAt], [createAt]), [processInstanceId]', $sql);
+        $this->assertCount(3, $this->query->calls[0]['bindings']);
+        $this->assertSame('PROC-TEST', $this->query->calls[0]['bindings'][0]);
+        $this->assertSame('RUNNING', json_decode(
+            (string) DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->where('process_instance_id', 'new-process')->value('workflow_payload'),
+            true,
+        )['status']);
+
+        $state = DB::connection('business_mirror')->table('business_mirror_table_states')->first();
+        $this->assertArrayHasKey('updateAt', json_decode((string) $state->high_watermark, true));
+    }
+
+    public function test_legacy_process_id_only_watermark_does_not_skip_new_processes(): void
+    {
+        $this->query->workflowBatches = [[
+            $this->workflow('existing-process', now()->subHour()->format('Y-m-d H:i:s.u')),
+        ]];
+        $this->synchronizer->sync(true, 10, 0);
+
+        DB::connection('business_mirror')->table('business_mirror_table_states')->update([
+            'high_watermark' => json_encode(['processInstanceId' => 'z-last-process']),
+        ]);
+        $this->query->calls = [];
+        $this->query->workflowBatches = [[
+            $this->workflow('a-new-process', now()->format('Y-m-d H:i:s.u')),
+        ]];
+
+        $this->synchronizer->sync(false, 10, 0);
+
+        $this->assertStringNotContainsString('AND [processInstanceId] > ?', $this->query->calls[0]['sql']);
+        $this->assertSame(2, DB::connection('business_mirror')->table('ding_payable_workflow_mirror')->count());
+    }
+
+    /** @return array<string,string> */
+    private function workflow(string $id, string $updateAt, string $status = 'COMPLETED'): array
+    {
+        return [
+            'processInstanceId' => $id,
+            'processCode' => 'PROC-TEST',
+            'updateAt' => $updateAt,
+            'createAt' => $updateAt,
+            'status' => $status,
+            'originatorUserId' => '',
+        ];
+    }
+}
+
+class FakeDingPayableWorkflowQuery extends UpstreamReadOnlyQuery
+{
+    /** @var list<list<array<string,mixed>>> */
+    public array $workflowBatches = [];
+
+    /** @var list<array{sql:string,bindings:array}> */
+    public array $calls = [];
+
+    public function __construct()
+    {
+        parent::__construct('operations_replica');
+    }
+
+    public function select(string $sql, array $bindings = []): array
+    {
+        $this->calls[] = compact('sql', 'bindings');
+
+        if (str_contains($sql, 'FROM [rpt].[ding_Workflow]')) {
+            return array_shift($this->workflowBatches) ?? [];
+        }
+
+        return [];
+    }
+}