182 lines
10 KiB
PHP
182 lines
10 KiB
PHP
<?php
|
|
|
|
use Tempo\Integration\WarehouseFeed;
|
|
use Tempo\Integration\WarehouseInbox;
|
|
use Tempo\Integration\WarehousePull;
|
|
|
|
require_once __DIR__.'/../Support/InboxFixture.php';
|
|
|
|
beforeEach(function () {
|
|
$this->db = inboxDatabase();
|
|
$this->inbox = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'warehouse-reader');
|
|
});
|
|
|
|
test('receiver persists original facts and checkpoint together and safely acknowledges an exact retry', function () {
|
|
expect($this->inbox->checkpoint())->toBe(['version' => 0, 'cursor' => null, 'last_event_id' => 0]);
|
|
$delivery = inboxEvent();
|
|
$receipt = inboxEvent(2, 'receive', $delivery['idempotency_key']);
|
|
$page = inboxPage([$delivery, $receipt]);
|
|
expect($this->inbox->accept($page, 0, null))->toMatchArray(['inserted' => 2, 'replayed' => false, 'version' => 1])
|
|
->and($this->inbox->fact($delivery['idempotency_key']))->toBe($delivery)
|
|
->and($this->inbox->fact($receipt['idempotency_key']))->toBe($receipt)
|
|
->and($this->inbox->checkpoint())->toBe(['version' => 1, 'cursor' => $page['next_cursor'], 'last_event_id' => 2]);
|
|
expect($this->inbox->accept($page, 0, null))->toMatchArray(['inserted' => 0, 'replayed' => true, 'version' => 1])
|
|
->and((int) $this->db->query('SELECT COUNT(*) FROM sgc_exchange_facts')->fetchColumn())->toBe(2);
|
|
$restarted = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'warehouse-reader');
|
|
expect($restarted->checkpoint())->toBe($this->inbox->checkpoint());
|
|
});
|
|
|
|
test('receiver rolls back an entire page and cursor when a later receipt lacks its delivery', function () {
|
|
$page = inboxPage([inboxEvent(), inboxEvent(3, 'receive', 'sgc:7:warehouse-event:2')]);
|
|
expect(fn () => $this->inbox->accept($page, 0, null))->toThrow(RuntimeException::class)
|
|
->and($this->inbox->checkpoint()['version'])->toBe(0)
|
|
->and($this->inbox->fact('sgc:7:warehouse-event:1'))->toBeNull();
|
|
$fixed = inboxPage([inboxEvent(), inboxEvent(2), inboxEvent(3, 'receive', 'sgc:7:warehouse-event:2')]);
|
|
expect($this->inbox->accept($fixed, 0, null)['inserted'])->toBe(3);
|
|
});
|
|
|
|
test('two readers cannot advance the same checkpoint with different pages', function () {
|
|
$second = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'warehouse-reader');
|
|
$before = $second->checkpoint();
|
|
$this->inbox->accept(inboxPage([inboxEvent()], 'cursor-a'), 0, null);
|
|
expect(fn () => $second->accept(inboxPage([inboxEvent(2)], 'cursor-b'), $before['version'], $before['cursor']))->toThrow(RuntimeException::class)
|
|
->and($this->inbox->checkpoint()['cursor'])->toBe('cursor-a')
|
|
->and($this->inbox->fact('sgc:7:warehouse-event:2'))->toBeNull();
|
|
});
|
|
|
|
test('receiver refuses stale pages mismatched request cursors and conflicting event identities', function () {
|
|
$this->inbox->accept(inboxPage([inboxEvent()]), 0, null);
|
|
expect(fn () => $this->inbox->accept(inboxPage([inboxEvent(2)]), 1, 'another-cursor'))->toThrow(RuntimeException::class)
|
|
->and(fn () => $this->inbox->accept(inboxPage([inboxEvent()]), 1, 'synthetic-opaque-cursor'))->toThrow(RuntimeException::class);
|
|
$otherStream = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'another-reader');
|
|
$conflict = inboxEvent();
|
|
$conflict['lines'][0]['quantity_milli'] = 2500;
|
|
expect(fn () => $otherStream->accept(inboxPage([inboxSign($conflict)]), 0, null))->toThrow(RuntimeException::class)
|
|
->and($otherStream->checkpoint()['version'])->toBe(0)
|
|
->and($this->inbox->fact('sgc:7:warehouse-event:1')['lines'][0]['quantity_milli'])->toBe(1500);
|
|
});
|
|
|
|
test('origins destination companies and consumer checkpoints remain separate', function () {
|
|
$page = inboxPage([inboxEvent()]);
|
|
$this->inbox->accept($page, 0, null);
|
|
foreach ([['synthetic-sgc-b', 7, 99], ['synthetic-sgc-a', 7, 100], ['synthetic-sgc-a', 8, 99]] as [$origin, $tenant, $destination]) {
|
|
$other = new WarehouseInbox($this->db, $origin, $tenant, $destination, 'warehouse-reader');
|
|
expect($other->fact('sgc:7:warehouse-event:1'))->toBeNull()->and($other->checkpoint()['version'])->toBe(0);
|
|
}
|
|
$otherStream = new WarehouseInbox($this->db, 'synthetic-sgc-a', 7, 99, 'another-reader');
|
|
expect($otherStream->accept($page, 0, null)['inserted'])->toBe(0)
|
|
->and($otherStream->checkpoint()['version'])->toBe(1);
|
|
});
|
|
|
|
test('receipt must match the exact source request and cannot be acknowledged twice', function () {
|
|
$delivery = inboxEvent();
|
|
$this->inbox->accept(inboxPage([$delivery], 'cursor-1'), 0, null);
|
|
$receipt = inboxEvent(2, 'receive', $delivery['idempotency_key']);
|
|
$receipt['request']['destination'] = 'Outro destino';
|
|
expect(fn () => $this->inbox->accept(inboxPage([inboxSign($receipt)]), 1, 'cursor-1'))->toThrow(RuntimeException::class);
|
|
$this->inbox->accept(inboxPage([inboxEvent(2, 'receive', $delivery['idempotency_key'])], 'cursor-2'), 1, 'cursor-1');
|
|
expect(fn () => $this->inbox->accept(inboxPage([inboxEvent(3, 'receive', $delivery['idempotency_key'])]), 2, 'cursor-2'))->toThrow(RuntimeException::class)
|
|
->and($this->inbox->checkpoint()['version'])->toBe(2);
|
|
});
|
|
|
|
test('empty polling pages remain resumable and do not invent a physical fact', function () {
|
|
$result = $this->inbox->accept(inboxPage([], 'empty-cursor'), 0, null);
|
|
expect($result['inserted'])->toBe(0)->and($result['has_more'])->toBeFalse()
|
|
->and($this->inbox->checkpoint())->toBe(['version' => 1, 'cursor' => 'empty-cursor', 'last_event_id' => 0]);
|
|
expect($this->inbox->accept(inboxPage([inboxEvent()]), 1, 'empty-cursor')['inserted'])->toBe(1);
|
|
});
|
|
|
|
test('malformed or altered feed data is refused before any receiving write', function () {
|
|
$cases = [];
|
|
foreach (['tenant_id' => 8, 'event_id' => '1', 'event_version' => 0, 'action' => 'authorize', 'occurred_at' => '2026-02-30T10:00:00+00:00', 'idempotency_key' => 'forged'] as $key => $value) {
|
|
$event = inboxEvent();
|
|
$event[$key] = $value;
|
|
$cases[] = inboxPage([inboxSign($event)]);
|
|
}
|
|
$event = inboxEvent();
|
|
$event['lines'][0]['quantity_milli'] = -1000;
|
|
$cases[] = inboxPage([inboxSign($event)]);
|
|
$event = inboxEvent();
|
|
$event['lines'][0]['quantity_milli'] = 1001;
|
|
$cases[] = inboxPage([$event]); // hash was not updated
|
|
$event = inboxEvent();
|
|
$event['lines'][0]['use'] = 'equipment';
|
|
$cases[] = inboxPage([inboxSign($event)]);
|
|
$cases[] = inboxPage([inboxEvent(2), inboxEvent()]);
|
|
$cases[] = inboxPage([inboxEvent(), inboxEvent()]);
|
|
$cases[] = inboxPage([], 'cursor', true);
|
|
$cases[] = ['schema' => 'unknown', 'items' => [], 'next_cursor' => 'cursor', 'has_more' => false];
|
|
foreach ($cases as $page) {
|
|
expect(fn () => $this->inbox->accept($page, 0, null))->toThrow(InvalidArgumentException::class);
|
|
}
|
|
expect((int) $this->db->query('SELECT COUNT(*) FROM sgc_exchange_streams')->fetchColumn())->toBe(0)
|
|
->and((int) $this->db->query('SELECT COUNT(*) FROM sgc_exchange_facts')->fetchColumn())->toBe(0);
|
|
});
|
|
|
|
test('new compatible metadata is preserved and incompatible schemas are rejected', function () {
|
|
$event = inboxEvent();
|
|
$event['extra_metadata'] = ['reference' => 'Origem adicional sintética'];
|
|
$event = inboxSign($event);
|
|
expect($this->inbox->accept(inboxPage([$event]), 0, null)['inserted'])->toBe(1)
|
|
->and($this->inbox->fact($event['idempotency_key'])['extra_metadata'])->toBe($event['extra_metadata']);
|
|
$bad = inboxPage([]);
|
|
$bad['schema'] = 'sgc.warehouse.fulfilment-feed.v2';
|
|
expect(fn () => WarehouseFeed::validate($bad, 7))->toThrow(InvalidArgumentException::class);
|
|
});
|
|
|
|
test('receiver never installs schema implicitly and refuses an externally owned transaction', function () {
|
|
$empty = new PDO('sqlite::memory:');
|
|
$inbox = new WarehouseInbox($empty, 'origin', 7, 99, 'reader');
|
|
expect((int) $empty->query("SELECT COUNT(*) FROM sqlite_master WHERE type='table'")->fetchColumn())->toBe(0);
|
|
$this->db->beginTransaction();
|
|
try {
|
|
expect(fn () => $this->inbox->accept(inboxPage([inboxEvent()]), 0, null))->toThrow(InvalidArgumentException::class)
|
|
->and($this->db->inTransaction())->toBeTrue();
|
|
} finally {
|
|
$this->db->rollBack();
|
|
}
|
|
});
|
|
|
|
test('bounded pull continues from the persisted cursor on the next execution', function () {
|
|
$seen = [];
|
|
$fetch = function (?string $cursor, int $size) use (&$seen): array {
|
|
$seen[] = [$cursor, $size];
|
|
|
|
return $cursor === null ? inboxPage([inboxEvent()], 'cursor-one', true)
|
|
: inboxPage([inboxEvent(2, 'receive', 'sgc:7:warehouse-event:1')], 'cursor-two');
|
|
};
|
|
$pull = new WarehousePull($this->inbox);
|
|
expect($pull->run($fetch, 1, 1))->toMatchArray(['pages' => 1, 'inserted' => 1, 'has_more' => true]);
|
|
expect($pull->run($fetch, 1, 1))->toMatchArray(['pages' => 1, 'inserted' => 1, 'has_more' => false])
|
|
->and($seen)->toBe([[null, 1], ['cursor-one', 1]])
|
|
->and($this->inbox->checkpoint()['last_event_id'])->toBe(2);
|
|
});
|
|
|
|
test('transport failure preserves already accepted pages and later resumes the failed page', function () {
|
|
$fetch = fn (?string $cursor, int $size) => $cursor === null ? inboxPage([inboxEvent()], 'cursor-one', true) : throw new RuntimeException('Indisponibilidade sintética.');
|
|
$pull = new WarehousePull($this->inbox);
|
|
expect(fn () => $pull->run($fetch, 5))->toThrow(RuntimeException::class)
|
|
->and($this->inbox->checkpoint())->toBe(['version' => 1, 'cursor' => 'cursor-one', 'last_event_id' => 1]);
|
|
$retry = function (?string $cursor, int $size): array {
|
|
expect($cursor)->toBe('cursor-one');
|
|
|
|
return inboxPage([inboxEvent(2, 'receive', 'sgc:7:warehouse-event:1')]);
|
|
};
|
|
expect($pull->run($retry, 5)['inserted'])->toBe(1);
|
|
});
|
|
|
|
test('pull stops at the actual end and rejects invalid bounds before calling a transport', function () {
|
|
$calls = 0;
|
|
$fetch = function (?string $cursor, int $size) use (&$calls): array {
|
|
$calls++;
|
|
|
|
return inboxPage([]);
|
|
};
|
|
$pull = new WarehousePull($this->inbox);
|
|
expect($pull->run($fetch, 100)['pages'])->toBe(1)->and($calls)->toBe(1);
|
|
foreach ([[0, 1], [101, 1], [1, 0], [1, 101]] as [$pages, $size]) {
|
|
expect(fn () => $pull->run($fetch, $pages, $size))->toThrow(InvalidArgumentException::class);
|
|
}
|
|
expect($calls)->toBe(1);
|
|
});
|