feat: rxjs (#29)

CodyTseng · Aug 18, 2024 · 6caf3af · 6caf3af
1 parent 645c7db
commit 6caf3af
Show file tree

Hide file tree

Showing 14 changed files with 217 additions and 153 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/packages/common/__test__/interfaces/event-repository.interface.spec.ts b/packages/common/__test__/interfaces/event-repository.interface.spec.ts
@@ -1,4 +1,5 @@
-import { EventRepository } from '../../src';
+import { EMPTY, from, Observable } from 'rxjs';
+import { EventRepository, Filter, toPromise } from '../../src';
 
 describe('EventRepository', () => {
   let eventRepository: EventRepository;
@@ -17,11 +18,7 @@ describe('EventRepository', () => {
       expect(await eventRepository.findOne({})).toBeNull();
       expect(eventRepository.find).toHaveBeenCalledWith({ limit: 1 });
 
-      eventRepository.find = jest.fn().mockResolvedValue([]);
-      expect(await eventRepository.findOne({})).toBeNull();
-      expect(eventRepository.find).toHaveBeenCalledWith({ limit: 1 });
-
-      eventRepository.find = jest.fn().mockReturnValue([]);
+      eventRepository.find = jest.fn().mockReturnValue(EMPTY);
       expect(await eventRepository.findOne({})).toBeNull();
       expect(eventRepository.find).toHaveBeenCalledWith({ limit: 1 });
     });
@@ -36,13 +33,34 @@ describe('EventRepository', () => {
       expect(await eventRepository.findOne({})).toEqual(event);
       expect(eventRepository.find).toHaveBeenCalledWith({ limit: 1 });
 
-      eventRepository.find = jest.fn().mockResolvedValue([event]);
+      eventRepository.find = jest.fn().mockReturnValue(from([event]));
       expect(await eventRepository.findOne({})).toEqual(event);
       expect(eventRepository.find).toHaveBeenCalledWith({ limit: 1 });
+    });
+  });
 
-      eventRepository.find = jest.fn().mockReturnValue([event]);
-      expect(await eventRepository.findOne({})).toEqual(event);
-      expect(eventRepository.find).toHaveBeenCalledWith({ limit: 1 });
+  describe('find$', () => {
+    it('should return find result', async () => {
+      const filter = {} as Filter;
+      const events = [{ id: 'a' }, { id: 'b' }];
+
+      eventRepository.find = jest.fn().mockReturnValue(events);
+      const obs1 = eventRepository.find$(filter);
+      expect(obs1 instanceof Observable).toBeTruthy();
+      expect(await toPromise(obs1)).toEqual(events);
+      expect(eventRepository.find).toHaveBeenCalledWith(filter);
+
+      eventRepository.find = jest.fn().mockResolvedValue(events);
+      const obs2 = eventRepository.find$(filter);
+      expect(obs2 instanceof Observable).toBeTruthy();
+      expect(await toPromise(obs2)).toEqual(events);
+      expect(eventRepository.find).toHaveBeenCalledWith(filter);
+
+      eventRepository.find = jest.fn().mockReturnValue(from(events));
+      const obs3 = eventRepository.find$(filter);
+      expect(obs3 instanceof Observable).toBeTruthy();
+      expect(await toPromise(obs3)).toEqual(events);
+      expect(eventRepository.find).toHaveBeenCalledWith(filter);
     });
   });
 });

diff --git a/packages/common/__test__/utils/event.spec.ts b/packages/common/__test__/utils/event.spec.ts
@@ -47,15 +47,15 @@ describe('EventUtils', () => {
 
     expect(
       EventUtils.validate(
-        createEvent({ created_at: getTimestampInSeconds() + 101 }),
+        createEvent({ created_at: getTimestampInSeconds() + 200 }),
         { createdAtUpperLimit: 100 },
       ),
     ).toBe(
       'invalid: created_at must not be later than 100 seconds from the current time',
     );
     expect(
       EventUtils.validate(
-        createEvent({ created_at: getTimestampInSeconds() - 101 }),
+        createEvent({ created_at: getTimestampInSeconds() - 200 }),
         { createdAtLowerLimit: 100 },
       ),
     ).toBe(

diff --git a/packages/common/package.json b/packages/common/package.json
@@ -29,6 +29,7 @@
   },
   "dependencies": {
     "@noble/curves": "^1.2.0",
-    "lru-cache": "^10.1.0"
+    "lru-cache": "^10.1.0",
+    "rxjs": "^7.8.1"
   }
 }
diff --git a/packages/common/src/interfaces/event-repository.interface.ts b/packages/common/src/interfaces/event-repository.interface.ts
@@ -1,3 +1,4 @@
+import { firstValueFrom, Observable } from 'rxjs';
 import { Event } from './event.interface';
 import { Filter } from './filter.interface';
 
@@ -40,7 +41,7 @@ export abstract class EventRepository {
    *
    * @param filter Query filter
    */
-  abstract find(filter: Filter): Promise<Event[]> | Event[];
+  abstract find(filter: Filter): Promise<Event[]> | Observable<Event> | Event[];
 
   /**
    * This method is called when the event repository should be closed. You can
@@ -55,7 +56,33 @@ export abstract class EventRepository {
    * @param filter Query filter
    */
   async findOne(filter: Filter): Promise<Event | null> {
-    const [event] = await this.find({ ...filter, limit: 1 });
+    const query = this.find({ ...filter, limit: 1 });
+    if (query instanceof Observable) {
+      return await firstValueFrom(query).catch(() => null);
+    }
+    const [event] = await query;
     return event ?? null;
   }
+
+  /**
+   * This method doesn't need to be implemented. It's just a helper method for
+   * transforming the `find` method to an observable.
+   *
+   * @param filter Query filter
+   */
+  find$(filter: Filter): Observable<Event> {
+    const query = this.find(filter);
+    if (query instanceof Observable) {
+      return query;
+    }
+    return new Observable(subscriber => {
+      (async (): Promise<void> => {
+        const events = await query;
+        for (const event of events) {
+          subscriber.next(event);
+        }
+        subscriber.complete();
+      })();
+    });
+  }
 }
diff --git a/packages/common/src/utils/index.ts b/packages/common/src/utils/index.ts
@@ -2,5 +2,6 @@ export * from './crypto.util';
 export * from './event.util';
 export * from './filter.util';
 export * from './proof-of-work.util';
-export * from './time.util';
+export * from './rxjs.util';
 export * from './shared.util';
+export * from './time.util';
diff --git a/packages/common/src/utils/rxjs.util.ts b/packages/common/src/utils/rxjs.util.ts
@@ -0,0 +1,12 @@
+import { Observable } from 'rxjs';
+
+export function toPromise<T>(observable: Observable<T>): Promise<T[]> {
+  return new Promise((resolve, reject) => {
+    const values: T[] = [];
+    observable.subscribe({
+      next: value => values.push(value),
+      error: reject,
+      complete: () => resolve(values),
+    });
+  });
+}
diff --git a/packages/core/__test__/nostr-relay.spec.ts b/packages/core/__test__/nostr-relay.spec.ts
@@ -1,3 +1,4 @@
+import { from } from 'rxjs';
 import {
   Client,
   ClientContext,
@@ -166,8 +167,8 @@ describe('NostrRelay', () => {
         .spyOn(nostrRelay['subscriptionService'], 'subscribe')
         .mockImplementation();
       const mockFind = jest
-        .spyOn(nostrRelay['eventService'], 'find')
-        .mockResolvedValue(events);
+        .spyOn(nostrRelay['eventService'], 'find$')
+        .mockReturnValue(from(events));
 
       const result = await nostrRelay.handleReqMessage(
         client,
@@ -230,8 +231,8 @@ describe('NostrRelay', () => {
         .spyOn(nostrRelay['subscriptionService'], 'subscribe')
         .mockImplementation();
       const mockFind = jest
-        .spyOn(nostrRelay['eventService'], 'find')
-        .mockResolvedValue(events);
+        .spyOn(nostrRelay['eventService'], 'find$')
+        .mockReturnValue(from(events));
 
       const result = await nostrRelay.handleReqMessage(
         client,
@@ -263,8 +264,8 @@ describe('NostrRelay', () => {
         .spyOn(nostrRelayWithoutHostname['subscriptionService'], 'subscribe')
         .mockImplementation();
       const mockFind = jest
-        .spyOn(nostrRelayWithoutHostname['eventService'], 'find')
-        .mockResolvedValue(events);
+        .spyOn(nostrRelayWithoutHostname['eventService'], 'find$')
+        .mockReturnValue(from(events));
 
       const result = await nostrRelayWithoutHostname.handleReqMessage(
         client,