fix: save event from mirrors (#376)

* fix: save event from mirrors

* docs: add config for mirroring event limits

* chore: add skip admission check (thanks YEGHRO)

* 2.0.1

* chore: bump some deps

.gitignore

@@ -35,4 +35,7 @@ dist
 *.env
 
 # Nostr data folder
-.nostr
+.nostr
+
+# Docker Compose overrides
+docker-compose.overrides.yml

CONFIGURATION.md

@@ -87,6 +87,8 @@ Running `nostream` for the first time creates the settings file in `<project_roo
 | paymentProcessors.lnurl.invoiceURL          | [LUD-06 Pay Request](https://github.com/lnurl/luds/blob/luds/06.md) provider URL. (e.g. https://getalby.com/lnurlp/your-username) |
 | mirroring.static[].address                  | Address of mirrored relay. (e.g. ws://100.100.100.100:8008) |
 | mirroring.static[].filters                  | Subscription filters used to mirror. |
+| mirroring.static[].limits.event                   | Event limit overrides for this mirror. See configurations under limits.event. |
+| mirroring.static[].skipAdmissionCheck       | Disable the admission fee check for events coming from this mirror. |
 | mirroring.static[].secret                   | Secret to pass to relays. Nostream relays only. Optional. |
 | workers.count                               | Number of workers to spin up to handle incoming connections. |
 |                                             | Spin workers as many CPUs are available when set to zero. Defaults to zero. |

Dockerfile

@@ -1,4 +1,4 @@
-FROM node:18-alpine3.16 as build
+FROM node:18-alpine3.16 AS build
 
 WORKDIR /build
 

docker-compose.yml

@@ -75,6 +75,7 @@ services:
     restart: on-failure
     networks:
       default:
+
   nostream-db:
     image: postgres
     container_name: nostream-db
@@ -96,6 +97,7 @@ services:
       timeout: 5s
       retries: 5
       start_period: 360s
+
   nostream-cache:
     image: redis:7.0.5-alpine3.16
     container_name: nostream-cache
@@ -110,6 +112,7 @@ services:
       interval: 1s
       timeout: 5s
       retries: 5
+
   nostream-migrate:
     image: node:18-alpine3.16
     container_name: nostream-migrate
@@ -131,7 +134,6 @@ services:
         condition: service_healthy
     networks:
       default:
-        ipv4_address: 10.10.10.254
 
 networks:
   default:

package-lock.json

@@ -1,16 +1,16 @@
 {
   "name": "nostream",
-  "version": "2.0.0",
+  "version": "2.0.1",
   "lockfileVersion": 2,
   "requires": true,
   "packages": {
     "": {
       "name": "nostream",
-      "version": "2.0.0",
+      "version": "2.0.1",
       "license": "MIT",
       "dependencies": {
         "@noble/secp256k1": "1.7.1",
-        "axios": "1.6.5",
+        "axios": "^1.7.7",
         "bech32": "2.0.0",
         "body-parser": "1.20.1",
         "debug": "4.3.4",
@@ -27,7 +27,7 @@
         "redis": "4.5.1",
         "rxjs": "7.8.0",
         "tor-control-ts": "^1.0.0",
-        "ws": "8.12.0"
+        "ws": "^8.18.0"
       },
       "devDependencies": {
         "@commitlint/cli": "17.2.0",
@@ -50,7 +50,7 @@
         "@types/ramda": "^0.28.13",
         "@types/sinon": "^10.0.11",
         "@types/sinon-chai": "^3.2.8",
-        "@types/ws": "^8.5.3",
+        "@types/ws": "^8.5.12",
         "@typescript-eslint/eslint-plugin": "^5.19.0",
         "@typescript-eslint/parser": "^5.19.0",
         "chai": "^4.3.6",
@@ -2385,9 +2385,9 @@
       "dev": true
     },
     "node_modules/@types/ws": {
-      "version": "8.5.3",
+      "version": "8.5.12",
-      "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.5.3.tgz",
+      "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.5.12.tgz",
-      "integrity": "sha512-6YOoWjruKj1uLf3INHH7D3qTXwFfEsg1kf3c0uDdSBJwfa/llkwIjrAGV7j7mVgGNbzTQ3HiHKKDXl6bJPD97w==",
+      "integrity": "sha512-3tPRkv1EtkDpzlgyKyI8pGsGZAGPEaXeu0DOj5DI25Ja91bdAYddYHbADRYVrZMRbfW+1l5YwXVDKohDJNQxkQ==",
       "dev": true,
       "dependencies": {
         "@types/node": "*"
@@ -2909,11 +2909,11 @@
       }
     },
     "node_modules/axios": {
-      "version": "1.6.5",
+      "version": "1.7.7",
-      "resolved": "https://registry.npmjs.org/axios/-/axios-1.6.5.tgz",
+      "resolved": "https://registry.npmjs.org/axios/-/axios-1.7.7.tgz",
-      "integrity": "sha512-Ii012v05KEVuUoFWmMW/UQv9aRIc3ZwkWDcM+h5Il8izZCtRVpDUfwpoFf7eOtajT3QiGR4yDUx7lPqHJULgbg==",
+      "integrity": "sha512-S4kL7XrjgBmvdGut0sN3yJxqYzrDOnivkBiN0OFs6hLiUam3UPvswUo0kqGyhqUZGEOytHyumEdXsAkgCOUf3Q==",
       "dependencies": {
-        "follow-redirects": "^1.15.4",
+        "follow-redirects": "^1.15.6",
         "form-data": "^4.0.0",
         "proxy-from-env": "^1.1.0"
       }
@@ -5095,9 +5095,9 @@
       "dev": true
     },
     "node_modules/follow-redirects": {
-      "version": "1.15.4",
+      "version": "1.15.9",
-      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.15.4.tgz",
+      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.15.9.tgz",
-      "integrity": "sha512-Cr4D/5wlrb0z9dgERpUL3LrmPKVDsETIJhaCMeDfuFYcqa5bldGV6wBsAN6X/vxlXQtFBMrXdXxdL8CbDTGniw==",
+      "integrity": "sha512-gew4GsXizNgdoRyqmyfMHyAmXsZDk6mHkSxZFCzW9gwlbtOW44CDtYavM+y+72qD/Vq2l550kMF52DT8fOLJqQ==",
       "funding": [
         {
           "type": "individual",
@@ -13444,9 +13444,9 @@
       }
     },
     "node_modules/ws": {
-      "version": "8.12.0",
+      "version": "8.18.0",
-      "resolved": "https://registry.npmjs.org/ws/-/ws-8.12.0.tgz",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.0.tgz",
-      "integrity": "sha512-kU62emKIdKVeEIOIKVegvqpXMSTAMLJozpHZaJNDYqBjzlSYXQGviYwN1osDLJ9av68qHd4a2oSjd7yD4pacig==",
+      "integrity": "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw==",
       "engines": {
         "node": ">=10.0.0"
       },
@@ -15477,9 +15477,9 @@
       "dev": true
     },
     "@types/ws": {
-      "version": "8.5.3",
+      "version": "8.5.12",
-      "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.5.3.tgz",
+      "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.5.12.tgz",
-      "integrity": "sha512-6YOoWjruKj1uLf3INHH7D3qTXwFfEsg1kf3c0uDdSBJwfa/llkwIjrAGV7j7mVgGNbzTQ3HiHKKDXl6bJPD97w==",
+      "integrity": "sha512-3tPRkv1EtkDpzlgyKyI8pGsGZAGPEaXeu0DOj5DI25Ja91bdAYddYHbADRYVrZMRbfW+1l5YwXVDKohDJNQxkQ==",
       "dev": true,
       "requires": {
         "@types/node": "*"
@@ -15841,11 +15841,11 @@
       "dev": true
     },
     "axios": {
-      "version": "1.6.5",
+      "version": "1.7.7",
-      "resolved": "https://registry.npmjs.org/axios/-/axios-1.6.5.tgz",
+      "resolved": "https://registry.npmjs.org/axios/-/axios-1.7.7.tgz",
-      "integrity": "sha512-Ii012v05KEVuUoFWmMW/UQv9aRIc3ZwkWDcM+h5Il8izZCtRVpDUfwpoFf7eOtajT3QiGR4yDUx7lPqHJULgbg==",
+      "integrity": "sha512-S4kL7XrjgBmvdGut0sN3yJxqYzrDOnivkBiN0OFs6hLiUam3UPvswUo0kqGyhqUZGEOytHyumEdXsAkgCOUf3Q==",
       "requires": {
-        "follow-redirects": "^1.15.4",
+        "follow-redirects": "^1.15.6",
         "form-data": "^4.0.0",
         "proxy-from-env": "^1.1.0"
       }
@@ -17529,9 +17529,9 @@
       "dev": true
     },
     "follow-redirects": {
-      "version": "1.15.4",
+      "version": "1.15.9",
-      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.15.4.tgz",
+      "resolved": "https://registry.npmjs.org/follow-redirects/-/follow-redirects-1.15.9.tgz",
-      "integrity": "sha512-Cr4D/5wlrb0z9dgERpUL3LrmPKVDsETIJhaCMeDfuFYcqa5bldGV6wBsAN6X/vxlXQtFBMrXdXxdL8CbDTGniw=="
+      "integrity": "sha512-gew4GsXizNgdoRyqmyfMHyAmXsZDk6mHkSxZFCzW9gwlbtOW44CDtYavM+y+72qD/Vq2l550kMF52DT8fOLJqQ=="
     },
     "foreground-child": {
       "version": "2.0.0",
@@ -23673,9 +23673,9 @@
       }
     },
     "ws": {
-      "version": "8.12.0",
+      "version": "8.18.0",
-      "resolved": "https://registry.npmjs.org/ws/-/ws-8.12.0.tgz",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.0.tgz",
-      "integrity": "sha512-kU62emKIdKVeEIOIKVegvqpXMSTAMLJozpHZaJNDYqBjzlSYXQGviYwN1osDLJ9av68qHd4a2oSjd7yD4pacig==",
+      "integrity": "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw==",
       "requires": {}
     },
     "xmlbuilder": {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "nostream",
-  "version": "2.0.0",
+  "version": "2.0.1",
   "description": "A Nostr relay written in Typescript.",
   "supportedNips": [
     1,
@@ -90,7 +90,7 @@
     "@types/ramda": "^0.28.13",
     "@types/sinon": "^10.0.11",
     "@types/sinon-chai": "^3.2.8",
-    "@types/ws": "^8.5.3",
+    "@types/ws": "^8.5.12",
     "@typescript-eslint/eslint-plugin": "^5.19.0",
     "@typescript-eslint/parser": "^5.19.0",
     "chai": "^4.3.6",
@@ -115,7 +115,7 @@
   },
   "dependencies": {
     "@noble/secp256k1": "1.7.1",
-    "axios": "1.6.5",
+    "axios": "^1.7.7",
     "bech32": "2.0.0",
     "body-parser": "1.20.1",
     "debug": "4.3.4",
@@ -132,7 +132,7 @@
     "redis": "4.5.1",
     "rxjs": "7.8.0",
     "tor-control-ts": "^1.0.0",
-    "ws": "8.12.0"
+    "ws": "^8.18.0"
   },
   "config": {
     "commitizen": {

src/@types/settings.ts

@@ -203,6 +203,10 @@ export interface Mirror {
   address: string
   filters?: SubscriptionFilter[]
   secret?: Secret
+  limits?: {
+    event?: EventLimits
+  }
+  skipAdmissionCheck?: boolean
 }
 
 export interface Mirroring {

src/app/static-mirroring-worker.ts

@@ -1,12 +1,15 @@
-import { anyPass, map, path } from 'ramda'
+import { anyPass, map, mergeDeepRight, path } from 'ramda'
 import { RawData, WebSocket } from 'ws'
 import cluster from 'cluster'
 import { randomUUID } from 'crypto'
 
 import { createRelayedEventMessage, createSubscriptionMessage } from '../utils/messages'
-import { isEventIdValid, isEventMatchingFilter, isEventSignatureValid } from '../utils/event'
+import { EventLimits, FeeSchedule, Mirror, Settings } from '../@types/settings'
-import { Mirror, Settings } from '../@types/settings'
+import { getEventExpiration, getEventProofOfWork, getPubkeyProofOfWork, getPublicKey, getRelayPrivateKey, isEventIdValid, isEventKindOrRangeMatch, isEventMatchingFilter, isEventSignatureValid, isExpiredEvent } from '../utils/event'
+import { IEventRepository, IUserRepository } from '../@types/repositories'
 import { createLogger } from '../factories/logger-factory'
+import { Event } from '../@types/event'
+import { EventExpirationTimeMetadataKey } from '../constants/base'
 import { IRunnable } from '../@types/base'
 import { OutgoingEventMessage } from '../@types/messages'
 import { RelayedEvent } from '../@types/event'
@@ -19,6 +22,8 @@ export class StaticMirroringWorker implements IRunnable {
   private config: Mirror
 
   public constructor(
+    private readonly eventRepository: IEventRepository,
+    private readonly userRepository: IUserRepository,
     private readonly process: NodeJS.Process,
     private readonly settings: () => Settings,
   ) {
@@ -57,7 +62,7 @@ export class StaticMirroringWorker implements IRunnable {
             this.send(JSON.stringify(createSubscriptionMessage(subscriptionId, filters)))
           }
         })
-        .on('message', async function (raw: RawData) {
+        .on('message', async (raw: RawData) => {
           try {
             const message = JSON.parse(raw.toString('utf8')) as OutgoingEventMessage
 
@@ -70,7 +75,7 @@ export class StaticMirroringWorker implements IRunnable {
               return
             }
 
-            const event = message[2]
+            let event = message[2]
 
             if (!anyPass(map(isEventMatchingFilter, config.filters))(event)) {
               return
@@ -80,10 +85,34 @@ export class StaticMirroringWorker implements IRunnable {
               return
             }
 
+            if (isExpiredEvent(event)) {
+              return
+            }
+
+            const eventExpiration = getEventExpiration(event)
+            if (eventExpiration) {
+              event = {
+                ...event,
+                [EventExpirationTimeMetadataKey]: eventExpiration,
+              } as any
+            }
+
+            if (!this.canAcceptEvent(event)) {
+              return
+            }
+
+            if (!await this.isUserAdmitted(event)) {
+              return
+            }
+
             since = Math.floor(Date.now() / 1000) - 30
 
-            if (cluster.isWorker && typeof process.send === 'function') {
+            debug('%s >> local: %s', config.address, event.id)
-              debug('%s >> local: %s', config.address, event.id)
+
+            const inserted = await this.eventRepository.create(event)
+
+            if (inserted && cluster.isWorker && typeof process.send === 'function') {
+
               process.send({
                 eventName: WebSocketServerAdapterEvent.Broadcast,
                 event,
@@ -110,6 +139,165 @@ export class StaticMirroringWorker implements IRunnable {
     this.client = createMirror(this.config)
   }
 
+  private getRelayPublicKey(): string {
+    const relayPrivkey = getRelayPrivateKey(this.settings().info.relay_url)
+    return getPublicKey(relayPrivkey)
+  }
+
+  private canAcceptEvent(event: Event): boolean {
+    if (this.getRelayPublicKey() === event.pubkey) {
+      debug(`event ${event.id} not accepted: pubkey is relay pubkey`)
+      return false
+    }
+
+    const now = Math.floor(Date.now() / 1000)
+
+    const eventLimits = this.settings().limits?.event ?? {}
+
+    const eventLimitOverrides = this.config.limits.event ?? {}
+
+    const limits = mergeDeepRight(eventLimits, eventLimitOverrides) as EventLimits
+
+    if (Array.isArray(limits.content)) {
+      for (const limit of limits.content) {
+        if (
+          typeof limit.maxLength !== 'undefined'
+          && limit.maxLength > 0
+          && event.content.length > limit.maxLength
+          && (
+            !Array.isArray(limit.kinds)
+            || limit.kinds.some(isEventKindOrRangeMatch(event))
+          )
+        ) {
+          debug(`event ${event.id} not accepted: content is longer than ${limit.maxLength} bytes`)
+          return false
+        }
+      }
+    } else if (
+      typeof limits.content?.maxLength !== 'undefined'
+      && limits.content?.maxLength > 0
+      && event.content.length > limits.content.maxLength
+      && (
+        !Array.isArray(limits.content.kinds)
+        || limits.content.kinds.some(isEventKindOrRangeMatch(event))
+      )
+    ) {
+      debug(`event ${event.id} not accepted: content is longer than ${limits.content.maxLength} bytes`)
+      return false
+    }
+
+    if (
+      typeof limits.createdAt?.maxPositiveDelta !== 'undefined'
+      && limits.createdAt.maxPositiveDelta > 0
+      && event.created_at > now + limits.createdAt.maxPositiveDelta) {
+      debug(`event ${event.id} not accepted: created_at is more than ${limits.createdAt.maxPositiveDelta} seconds in the future`)
+      return false
+    }
+
+    if (
+      typeof limits.createdAt?.maxNegativeDelta !== 'undefined'
+      && limits.createdAt.maxNegativeDelta > 0
+      && event.created_at < now - limits.createdAt.maxNegativeDelta) {
+      debug(`event ${event.id} not accepted: created_at is more than ${limits.createdAt.maxNegativeDelta} seconds in the past`)
+      return false
+    }
+
+    if (
+      typeof limits.eventId?.minLeadingZeroBits !== 'undefined'
+      && limits.eventId.minLeadingZeroBits > 0
+    ) {
+      const pow = getEventProofOfWork(event.id)
+      if (pow < limits.eventId.minLeadingZeroBits) {
+        debug(`event ${event.id} not accepted: pow difficulty ${pow}<${limits.eventId.minLeadingZeroBits}`)
+        return false
+      }
+    }
+
+    if (
+      typeof limits.pubkey?.minLeadingZeroBits !== 'undefined'
+      && limits.pubkey.minLeadingZeroBits > 0
+    ) {
+      const pow = getPubkeyProofOfWork(event.pubkey)
+      if (pow < limits.pubkey.minLeadingZeroBits) {
+        debug(`event ${event.id} not accepted: pow pubkey difficulty ${pow}<${limits.pubkey.minLeadingZeroBits}`)
+        return false
+      }
+    }
+
+    if (
+      typeof limits.pubkey?.whitelist !== 'undefined'
+      && limits.pubkey.whitelist.length > 0
+      && !limits.pubkey.whitelist.some((prefix) => event.pubkey.startsWith(prefix))
+    ) {
+      debug(`event ${event.id} not accepted: pubkey not allowed: ${event.pubkey}`)
+      return false
+    }
+
+    if (
+      typeof limits.pubkey?.blacklist !== 'undefined'
+      && limits.pubkey.blacklist.length > 0
+      && limits.pubkey.blacklist.some((prefix) => event.pubkey.startsWith(prefix))
+    ) {
+      debug(`event ${event.id} not accepted: pubkey not allowed: ${event.pubkey}`)
+      return false
+    }
+
+    if (
+      typeof limits.kind?.whitelist !== 'undefined'
+      && limits.kind.whitelist.length > 0
+      && !limits.kind.whitelist.some(isEventKindOrRangeMatch(event))) {
+      debug(`blocked: event kind ${event.kind} not allowed`)
+      return false
+    }
+
+    if (
+      typeof limits.kind?.blacklist !== 'undefined'
+      && limits.kind.blacklist.length > 0
+      && limits.kind.blacklist.some(isEventKindOrRangeMatch(event))) {
+      debug(`blocked: event kind ${event.kind} not allowed`)
+      return false
+    }
+
+    return true
+  }
+
+  protected async isUserAdmitted(event: Event): Promise<boolean> {
+    const currentSettings = this.settings()
+
+    if (this.config.skipAdmissionCheck === true) {
+      return true
+    }
+
+    if (currentSettings.payments?.enabled !== true) {
+      return true
+    }
+
+    const isApplicableFee = (feeSchedule: FeeSchedule) =>
+      feeSchedule.enabled
+      && !feeSchedule.whitelists?.pubkeys?.some((prefix) => event.pubkey.startsWith(prefix))
+      && !feeSchedule.whitelists?.event_kinds?.some(isEventKindOrRangeMatch(event))
+
+    const feeSchedules = currentSettings.payments?.feeSchedules?.admission?.filter(isApplicableFee)
+
+    if (!Array.isArray(feeSchedules) || !feeSchedules.length) {
+      return true
+    }
+
+    const user = await this.userRepository.findByPubkey(event.pubkey)
+    if (user?.isAdmitted !== true) {
+      debug(`user not admitted: ${event.pubkey}`)
+      return false
+    }
+
+    const minBalance = currentSettings.limits?.event?.pubkey?.minBalance
+    if (minBalance && user.balance < minBalance) {
+      debug(`user not admitted: user balance ${user.balance} < ${minBalance}`)
+      return false
+    }
+
+    return true
+  }
+
   private onMessage(message: { eventName: string, event: unknown, source: string }): void {
     if (
       message.eventName !== WebSocketServerAdapterEvent.Broadcast

src/factories/static-mirroring.worker-factory.ts

@@ -1,6 +1,19 @@
+import { getMasterDbClient, getReadReplicaDbClient } from '../database/client'
 import { createSettings } from './settings-factory'
+import { EventRepository } from '../repositories/event-repository'
 import { StaticMirroringWorker } from '../app/static-mirroring-worker'
+import { UserRepository } from '../repositories/user-repository'
 
 export const staticMirroringWorkerFactory = () => {
-  return new StaticMirroringWorker(process, createSettings)
+  const dbClient = getMasterDbClient()
+  const readReplicaDbClient = getReadReplicaDbClient()
+  const eventRepository = new EventRepository(dbClient, readReplicaDbClient)
+  const userRepository = new UserRepository(dbClient)
+
+  return new StaticMirroringWorker(
+    eventRepository,
+    userRepository,
+    process,
+    createSettings,
+  )
 }

src/utils/settings.ts

@@ -125,7 +125,7 @@ export class SettingsStatic {
     const settingsFilePath = join(basePath, `settings.${fileType}`)
 
     const reload = () => {
-      console.log('reloading settings')
+      debug('reloading settings')
       SettingsStatic._settings = undefined
       SettingsStatic.createSettings()
     }