Activities/Wallet Refactor

2025-04-18 05:17:20 +00:00
parent 408d3b8d8c
commit 7e4508cf5a
18 changed files with 706 additions and 1111 deletions
@@ -0,0 +1,275 @@
+import { BaseStore } from './base-store.js';
+import { StoreComposite } from './store-composite.js';
+
+import { UUIDV7 } from '../utils';
+
+// Make the given property in a type optional.
+// TODO: This is an extremely useful utility. If we use it elsewhere, move to own "type-utils.ts" file.
+type Optional<T, K extends keyof T> = Omit<T, K> & Partial<Pick<T, K>>;
+
+// Base Activity interface
+export interface Activity {
+  entityId: string;
+  name: string;
+  value: unknown;
+  timestamp: bigint;
+}
+
+// Define the snapshot structure
+export type SnapshotByType = {
+  [type: string]: {
+    [id: string]: Activity;
+  };
+};
+
+// Policy type definition
+export type PolicyFunction = (
+  existingActivity: Activity | undefined,
+  newActivity: Activity,
+) => Activity;
+
+/**
+ * Implements an event sourcing pattern for managing application state
+ * with a simplified interface focused on the current snapshot.
+ *
+ * @example
+ * ```typescript
+ * // Create Activities manager with a store implementation
+ * const activities = await Activities.fromLocalStore(localStore);
+ *
+ * // Register a custom policy for wallet name conflicts
+ * activities.registerPolicy('WalletName', (existing, newActivity) => {
+ *   // Keep the most recent wallet name
+ *   return existing?.timestamp > newActivity.timestamp ? existing : newActivity;
+ * });
+ *
+ * // Add a new wallet name activity
+ * const walletName: WalletNameInput = {
+ *   name: 'WalletName',
+ *   value: 'Test Wallet'
+ * };
+ * const savedActivity = await activities.add(walletName);
+ * ```
+ */
+export class Activities {
+  public snapshot: SnapshotByType = {};
+  private policies = new Map<string, PolicyFunction>();
+  private storeKeyMap = new Map<string, string>();
+
+  /**
+   * Create an Activities instance from the given composite store.
+   */
+  static async from(compositeStore: StoreComposite) {
+    const activities = new Activities(compositeStore);
+    await activities.compileSnapshot();
+    return activities;
+  }
+
+  /**
+   * Create an Activities instance from the given primary and secondary stores
+   */
+  static async fromStores(
+    primaryStore: BaseStore,
+    secondaryStores?: Array<BaseStore>,
+  ): Promise<Activities> {
+    const compositeStore = await StoreComposite.from(
+      primaryStore,
+      secondaryStores,
+    );
+    return Activities.from(compositeStore);
+  }
+
+  private constructor(public store: StoreComposite) {
+    this.store.on(
+      'set',
+      async (payload: { key: string; value: Activity | null }) => {
+        if (payload.value !== null) {
+          await this.processActivity(payload.key, payload.value);
+        }
+      },
+    );
+  }
+
+  /**
+   * Add a new activity to the store
+   * @param activity The activity to add (id and timestamp are optional)
+   * @returns The stored activity with all fields populated
+   */
+  public async add<T extends Optional<Activity, 'timestamp'>>(
+    activity: T,
+  ): Promise<Activity> {
+    // Generate timestamp if not provided
+    const timestamp = activity.timestamp ?? BigInt(Date.now());
+
+    // Generate a UUID based on the timestamp
+    const key = UUIDV7.fromTimestamp(timestamp).toString();
+
+    // Generate an ID if not provided
+    const id = activity.entityId ?? key;
+
+    const completeActivity: Activity = {
+      name: activity.name,
+      value: activity.value,
+      entityId: id,
+      timestamp,
+    };
+
+    await this.store.set(key, completeActivity);
+
+    return completeActivity;
+  }
+
+  /**
+   * Get a specific activity by type and id with type inference
+   */
+  public get<T extends Activity>(type: T['name'], id: string): T | undefined {
+    const activity = this.snapshot[type]?.[id];
+    return activity as T | undefined;
+  }
+
+  /**
+   * Get all activities of a specific type
+   */
+  public getByType<T extends Activity>(type: T['name']): T[] {
+    const activities = this.snapshot[type];
+    if (!activities) {
+      return [];
+    }
+    return Object.values(activities) as T[];
+  }
+
+  /**
+   * Register a policy for handling conflicts between activities of the same type
+   * @param type Activity type to register policy for
+   * @param policyFn Function that decides which activity to keep
+   */
+  public registerPolicy(type: string, policyFn: PolicyFunction): void {
+    this.policies.set(type, policyFn);
+  }
+
+  /**
+   * Process a single activity and update the snapshot
+   */
+  private async processActivity(
+    storeKey: string,
+    activity: Activity,
+  ): Promise<void> {
+    if (!activity) {
+      return;
+    }
+
+    // Ensure timestamp exists by extracting from UUID if needed
+    if (!activity.timestamp) {
+      activity.timestamp = UUIDV7.from(storeKey).getTimestamp();
+    }
+
+    // Ensure ID exists
+    if (!activity.entityId) {
+      activity.entityId = storeKey;
+    }
+
+    const snapshotKey = this.getSnapshotKey(activity);
+    const existingActivity = this.getActivityFromSnapshot(
+      activity.name,
+      activity.entityId,
+    );
+
+    // Apply policy to determine which activity to keep
+    const keptActivity = this.applyPolicy(existingActivity, activity);
+
+    // Update snapshot with the kept activity
+    this.setActivityInSnapshot(keptActivity);
+
+    if (keptActivity === activity) {
+      // If we kept the new activity, update the store key mapping
+      const oldStoreKey = this.storeKeyMap.get(snapshotKey);
+      if (oldStoreKey && oldStoreKey !== storeKey) {
+        // Delete the old activity from the store
+        await this.store.set(oldStoreKey, null);
+      }
+      this.storeKeyMap.set(snapshotKey, storeKey);
+    } else if (storeKey !== this.storeKeyMap.get(snapshotKey)) {
+      // If we kept the existing activity, delete the new activity from the store
+      await this.store.set(storeKey, null);
+    }
+  }
+
+  /**
+   * Compile a full snapshot from scratch
+   * This happens automatically on initialization
+   */
+  private async compileSnapshot(): Promise<SnapshotByType> {
+    // Clear existing snapshot
+    this.snapshot = {};
+    this.storeKeyMap.clear();
+
+    // Get all activity data from our store
+    const activities = await this.store.all<Activity>();
+    const entries = Object.entries(activities);
+
+    // Filter for null payloads
+    const filteredEntries = entries.filter(([_, payload]) => payload);
+
+    // Sort activities by timestamp using UUID v7's embedded timestamp
+    const activitiesSorted = filteredEntries.sort(([keyA], [keyB]) => {
+      const timestampA = UUIDV7.from(keyA).getTimestamp();
+      const timestampB = UUIDV7.from(keyB).getTimestamp();
+      return Number(timestampA - timestampB);
+    });
+
+    // Process each activity in order
+    for (const [storeKey, activity] of activitiesSorted) {
+      await this.processActivity(storeKey, activity);
+    }
+
+    return this.snapshot;
+  }
+
+  /**
+   * Get an activity from the snapshot by type and id
+   */
+  private getActivityFromSnapshot(
+    type: string,
+    id: string,
+  ): Activity | undefined {
+    return this.snapshot[type]?.[id];
+  }
+
+  /**
+   * Set an activity in the snapshot, creating necessary object structure
+   */
+  private setActivityInSnapshot(activity: Activity): void {
+    if (!this.snapshot[activity.name]) {
+      this.snapshot[activity.name] = {};
+    }
+
+    this.snapshot[activity.name][activity.entityId] = activity;
+  }
+
+  /**
+   * Get the snapshot key for an activity
+   */
+  private getSnapshotKey(activity: Activity): string {
+    return `${activity.name}:${activity.entityId}`;
+  }
+
+  /**
+   * Apply policy to determine which activity should be kept
+   */
+  private applyPolicy(
+    existingActivity: Activity | undefined,
+    newActivity: Activity,
+  ): Activity {
+    const policy = this.policies.get(newActivity.name);
+
+    if (!policy) {
+      // Default policy: compare timestamps
+      if (!existingActivity) return newActivity;
+      return existingActivity.timestamp >= newActivity.timestamp
+        ? existingActivity
+        : newActivity;
+    }
+
+    return policy(existingActivity, newActivity);
+  }
+}
@@ -1,3 +1,4 @@
+export * from './activities.js';
 export * from './base-store.js';
 export * from './store-composite.js';
 export * from './store-inmemory.js';
@@ -11,7 +11,10 @@ import { BaseStore } from './base-store.js';
 * UUIDv7 keys are used to ensure uniqueness of keys.
 */
 export class StoreComposite extends BaseStore {
-  static async from(localStore: BaseStore, remoteStores: Array<BaseStore>) {
+  static async from(
+    localStore: BaseStore,
+    remoteStores: Array<BaseStore> = [],
+  ) {
    const compositeStore = new this(localStore, remoteStores);
    await compositeStore.sync();
    return compositeStore;
@@ -19,55 +22,9 @@ export class StoreComposite extends BaseStore {

  constructor(
    private localStore: BaseStore,
-    private remoteStores: Array<BaseStore>,
+    private remoteStores: Array<BaseStore> = [],
  ) {
    super();
-
-    // Local store events propagate to remote stores
-    this.localStore.on('set', async (payload) => {
-      // Propagate to all remote stores in parallel
-      await Promise.all(
-        this.remoteStores.map((remoteStore) =>
-          remoteStore.set(payload.key, payload.value),
-        ),
-      );
-    });
-
-    // Local store events propagate to remote stores
-    this.localStore.on('delete', async (payload) => {
-      // Propagate to all remote stores in parallel
-      await Promise.all(
-        this.remoteStores.map((remoteStore) => remoteStore.delete(payload.key)),
-      );
-    });
-
-    // Local store events propagate to remote stores
-    this.localStore.on('clear', async () => {
-      // Propagate to all remote stores in parallel
-      await Promise.all(
-        this.remoteStores.map((remoteStore) => remoteStore.clear()),
-      );
-    });
-
-    // Remote store events propagate to local store
-    for (const remoteStore of this.remoteStores) {
-      remoteStore.on('set', async (payload) => {
-        // Skip if the local store already has the key
-        // NOTE: It's important we DO NOT repropagate events that have already occurred.
-        //       If we do, we can end up in infinite loops.
-        if (await this.localStore.has(payload.key)) return;
-
-        await this.localStore.set(payload.key, payload.value);
-      });
-
-      remoteStore.on('delete', async (payload) => {
-        // NOTE: It's important we DO NOT repropagate events that have already occurred.
-        //       If we do, we can end up in infinite loops.
-        if (await this.localStore.has(payload.key)) {
-          await this.localStore.delete(payload.key);
-        }
-      });
-    }
  }

  async all<T>() {
@@ -108,43 +65,74 @@ export class StoreComposite extends BaseStore {
    ]);
  }

+  /**
+   * Add a new remote store and set up synchronization
+   */
+  public async addRemoteStore(remoteStore: BaseStore): Promise<void> {
+    this.remoteStores.push(remoteStore);
+
+    // Propagate existing local events to the new remote
+    this.localStore.on('set', async (payload) => {
+      await remoteStore.set(payload.key, payload.value);
+    });
+
+    this.localStore.on('delete', async (payload) => {
+      await remoteStore.delete(payload.key);
+    });
+
+    this.localStore.on('clear', async () => {
+      await remoteStore.clear();
+    });
+
+    // Set up remote to local propagation
+    remoteStore.on('set', async (payload) => {
+      if (await this.localStore.has(payload.key)) return;
+      await this.localStore.set(payload.key, payload.value);
+    });
+
+    remoteStore.on('delete', async (payload) => {
+      if (await this.localStore.has(payload.key)) {
+        await this.localStore.delete(payload.key);
+      }
+    });
+
+    remoteStore.on('clear', async () => {
+      await this.localStore.clear();
+    });
+
+    // Sync with the new remote
+    await this.syncWithRemote(remoteStore);
+  }
+
+  private async syncWithRemote(remoteStore: BaseStore): Promise<void> {
+    const remoteKeys = await remoteStore.keys();
+    const localKeys = await this.localStore.keys();
+
+    const missingLocalKeys = remoteKeys.filter(
+      (key) => !localKeys.includes(key),
+    );
+    const missingRemoteKeys = localKeys.filter(
+      (key) => !remoteKeys.includes(key),
+    );
+
+    await Promise.all([
+      ...missingLocalKeys.map(async (key) => {
+        const value = await remoteStore.get(key);
+        await this.localStore.set(key, value);
+      }),
+      ...missingRemoteKeys.map(async (key) => {
+        const value = await this.localStore.get(key);
+        await remoteStore.set(key, value);
+      }),
+    ]);
+  }
+
  async sync(): Promise<void> {
    // Sync all remote stores in parallel
    await Promise.all(
-      this.remoteStores.map(async (remoteStore) => {
-        // TODO: We want to negotiate using an Inverse Bloom Filter first to optimize the sync.
-        //       Currently, this is doing a naive set comparison.
-
-        // Get all keys from the remote store
-        const remoteKeys = await remoteStore.keys();
-        const localKeys = await this.localStore.keys();
-
-        // Find keys missing in local store
-        const missingLocalKeys = remoteKeys.filter(
-          (key) => !localKeys.includes(key),
-        );
-
-        // Find keys missing in remote store
-        const missingRemoteKeys = localKeys.filter(
-          (key) => !remoteKeys.includes(key),
-        );
-
-        // Sync missing keys from remote to local (in parallel)
-        await Promise.all(
-          missingLocalKeys.map(async (key) => {
-            const value = await remoteStore.get(key);
-            await this.localStore.set(key, value);
-          }),
-        );
-
-        // Sync missing keys from local to remote (in parallel)
-        await Promise.all(
-          missingRemoteKeys.map(async (key) => {
-            const value = await this.localStore.get(key);
-            await remoteStore.set(key, value);
-          }),
-        );
-      }),
+      this.remoteStores.map(async (remoteStore) =>
+        this.syncWithRemote(remoteStore),
+      ),
    );
  }
 }
@@ -7,7 +7,12 @@ import { BaseStore } from './base-store.js';
 * This implementation can be used testing and caching of expensive operations.
 */
 export class StoreInMemory extends BaseStore {
-  private store: Map<string, any>;
+  // TODO: Eventually this amy accept indexes as an argument.
+  static from(): StoreInMemory {
+    return new StoreInMemory();
+  }
+
+  private store: Map<string, unknown>;

  constructor() {
    super();