refactor: staged writes in private events

yarn-project/pxe/src/contract_function_simulator/oracle/utility_execution_oracle.ts

@@ -409,7 +409,7 @@ export class UtilityExecutionOracle implements IMiscOracle, IUtilityExecutionOra
       ),
     );
 
-    const eventService = new EventService(this.anchorBlockStore, this.aztecNode, this.privateEventStore);
+    const eventService = new EventService(this.anchorBlockStore, this.aztecNode, this.privateEventStore, this.jobId);
     const eventDeliveries = eventValidationRequests.map(request =>
       eventService.deliverEvent(
         request.contractAddress,

yarn-project/pxe/src/events/event_service.test.ts

@@ -91,7 +91,7 @@ describe('deliverEvent', () => {
       ]),
     );
 
-    eventService = new EventService(anchorBlockStore, aztecNode, privateEventStore);
+    eventService = new EventService(anchorBlockStore, aztecNode, privateEventStore, 'test');
   });
 
   function runDeliverEvent(
@@ -141,12 +141,16 @@ describe('deliverEvent', () => {
     await runDeliverEvent();
 
     // I should be able to retrieve the private event I just saved using getPrivateEvents
-    const result = await privateEventStore.getPrivateEvents(eventSelector, {
-      contractAddress,
-      fromBlock: blockNumber,
-      toBlock: blockNumber + 1,
-      scopes: [recipient],
-    });
+    const result = await privateEventStore.getPrivateEvents(
+      eventSelector,
+      {
+        contractAddress,
+        fromBlock: blockNumber,
+        toBlock: blockNumber + 1,
+        scopes: [recipient],
+      },
+      'test',
+    );
 
     expect(result.length).toEqual(1);
     expect(result[0].packedEvent).toEqual(eventContent);

yarn-project/pxe/src/events/event_service.ts

@@ -14,6 +14,7 @@ export class EventService {
     private readonly anchorBlockStore: AnchorBlockStore,
     private readonly aztecNode: AztecNode,
     private readonly privateEventStore: PrivateEventStore,
+    private readonly jobId: string,
   ) {}
 
   public async deliverEvent(
@@ -74,6 +75,7 @@ export class EventService {
         l2BlockNumber: nullifierIndex.l2BlockNumber, // Block number in which the event was emitted
         l2BlockHash: nullifierIndex.l2BlockHash, // Block hash in which the event was emitted
       },
+      this.jobId,
     );
   }
 }

yarn-project/pxe/src/pxe.test.ts

@@ -220,13 +220,20 @@ describe('PXE', () => {
 
       const randomness = Fr.random();
 
-      await privateEventStore.storePrivateEventLog(eventSelector, randomness, event.packedEvent, eventIndex++, {
-        contractAddress,
-        scope,
-        txHash: event.txHash,
-        l2BlockNumber: event.l2BlockNumber,
-        l2BlockHash: event.l2BlockHash,
-      });
+      await privateEventStore.storePrivateEventLog(
+        eventSelector,
+        randomness,
+        event.packedEvent,
+        eventIndex++,
+        {
+          contractAddress,
+          scope,
+          txHash: event.txHash,
+          l2BlockNumber: event.l2BlockNumber,
+          l2BlockHash: event.l2BlockHash,
+        },
+        'test',
+      );
 
       return event;
     }
@@ -235,6 +242,7 @@ describe('PXE', () => {
       // Store a couple of events to exercise `getPrivateEvents`
       const event1 = await storeEvent();
       const event2 = await storeEvent();
+      await privateEventStore.commit('test');
 
       const events = await pxe.getPrivateEvents(eventSelector, {
         contractAddress,
@@ -275,6 +283,8 @@ describe('PXE', () => {
           storeEvent(lastKnownBlockNumber + 1),
           storeEvent(lastKnownBlockNumber + 1),
         ]);
+
+        await privateEventStore.commit('test');
       });
 
       it('filters by txHash', async () => {

yarn-project/pxe/src/pxe.ts

@@ -157,7 +157,7 @@ export class PXE {
     );
 
     const jobCoordinator = new JobCoordinator(store);
-    jobCoordinator.registerStores([capsuleStore, senderTaggingStore, recipientTaggingStore]);
+    jobCoordinator.registerStores([capsuleStore, senderTaggingStore, recipientTaggingStore, privateEventStore]);
 
     const debugUtils = new PXEDebugUtils(contractStore, noteStore);
 
@@ -1099,7 +1099,7 @@ export class PXE {
         `Getting private events for ${sanitizedFilter.contractAddress.toString()} from ${sanitizedFilter.fromBlock} to ${sanitizedFilter.toBlock}`,
       );
 
-      return this.privateEventStore.getPrivateEvents(eventSelector, sanitizedFilter);
+      return this.privateEventStore.getPrivateEvents(eventSelector, sanitizedFilter, jobId);
     });
   }