import StreamID , { CommitID } from '@ceramicnetwork/streamid' ; import { AnchorService , AnchorStatus , Context , CreateOpts , LoadOpts , PinningOpts , PublishOpts , StreamState , SyncOptions , UpdateOpts , } from '@ceramicnetwork/common' ; import { PinStore } from '../store/pin-store' ; import { DiagnosticsLogger } from '@ceramicnetwork/common' ; import { ExecutionQueue } from './execution-queue' ; import { RunningState } from './running-state' ; import { StateManager } from './state-manager' ; import type { Dispatcher } from '../dispatcher' ; import type { ConflictResolution } from '../conflict-resolution' ; import type { HandlersMap } from '../handlers-map' ; import type { StateValidation } from './state-validation' ; import { Observable } from 'rxjs' ; import { StateCache } from './state-cache' ; import { SnapshotState } from './snapshot-state' ; import Utils from '../utils' ; export type RepositoryDependencies = { dispatcher : Dispatcher ; pinStore : PinStore ; context : Context ; handlers : HandlersMap ; anchorService : AnchorService ; conflictResolution : ConflictResolution ; stateValidation : StateValidation ; } ; const DEFAULT_LOAD_OPTS = { sync : SyncOptions . PREFER_CACHE , syncTimeoutSeconds : 3 } ; export class Repository { /**
     * Serialize loading operations per streamId.
     */ readonly loadingQ : ExecutionQueue ; /**
     * Serialize operations on state per streamId.
     */ readonly executionQ : ExecutionQueue ; /**
     * In-memory cache of the currently running streams.
     */ readonly inmemory : StateCache < RunningState > ; /**
     * Various dependencies.
     */ #deps : RepositoryDependencies ; /**
     * Instance of StateManager for performing operations on stream state.
     */ stateManager : StateManager ; param cacheLimit param logger param concurrencyLimit constructor ( cacheLimit , concurrencyLimit , private readonly logger ) { this . loadingQ = new ExecutionQueue ( concurrencyLimit , logger ) ; this . executionQ = new ExecutionQueue ( concurrencyLimit , logger ) ; this . inmemory = new StateCache ( cacheLimit , ( state$ ) => state$ . complete ( ) ) ; this . updates$ = this . updates$ . bind ( this ) ; } get pinStore ( ) : PinStore { return this . #deps . pinStore ; } setDeps ( deps ) { this . #deps = deps ; this . stateManager = new StateManager ( deps . dispatcher , deps . pinStore , this . executionQ , deps . anchorService , deps . conflictResolution , this . logger , ( streamId ) => this . get ( streamId ) , ( streamId , opts ) => this . load ( streamId , opts ) ) ; } private fromMemory ( streamId ) { return this . inmemory . get ( streamId . toString ( ) ) ; } private async fromStateStore ( streamId ) { const streamState = await this . #deps . pinStore . stateStore . load ( streamId ) ; if ( streamState ) { const runningState = new RunningState ( streamState , true ) ; this . add ( runningState ) ; const toRecover = runningState . value . anchorStatus === AnchorStatus . PENDING || runningState . value . anchorStatus === AnchorStatus . PROCESSING ; if ( toRecover && this . stateManager . anchorService ) { this . stateManager . confirmAnchorResponse ( runningState ) ; } return runningState ; } else { return undefined ; } } private async fromNetwork ( streamId ) { const handler = this . #deps . handlers . get ( streamId . typeName ) ; const genesisCid = streamId . cid ; const commitData = await Utils . getCommitData ( this . #deps . dispatcher , genesisCid ) ; if ( commitData == null ) { throw new Error ( `No genesis commit found with CID ${ genesisCid . toString ( ) }` ) ; } commitData . disableTimecheck = true ; const state = await handler . applyCommit ( commitData , this . #deps . context ) ; await this . #deps . stateValidation . validate ( state , state . content ) ; const state$ = new RunningState ( state , false ) ; this . add ( state$ ) ; this . logger . verbose ( `Genesis commit for stream ${ streamId . toString ( ) } successfully loaded` ) ; return state$ ; } param streamId returns async _loadGenesis ( streamId ) { let stream = this . fromMemory ( streamId ) ; if ( stream ) { return [ stream , true ] ; } stream = await this . fromStateStore ( streamId ) ; if ( stream ) { return [ stream , this . stateManager . wasPinnedStreamSynced ( streamId ) ] ; } stream = await this . fromNetwork ( streamId ) ; return [ stream , false ] ; } /**
     * Returns a stream from wherever we can get information about it.
     * Starts by checking if the stream state is present in the in-memory cache, if not then
     * checks the state store, and finally loads the stream from pubsub.
     */ async load ( streamId , opts ) { opts = { ... DEFAULT_LOAD_OPTS , ... opts } ; const state$ = await this . loadingQ . forStream ( streamId ) . run ( async ( ) => { const [ stream , synced ] = await this . _loadGenesis ( streamId ) ; if ( opts . sync == SyncOptions . PREFER_CACHE && synced ) { return stream ; } if ( opts . sync == SyncOptions . NEVER_SYNC ) { return this . stateManager . verifyLoneGenesis ( stream ) ; } await this . stateManager . sync ( stream , opts . syncTimeoutSeconds * 1000 ) ; return this . stateManager . verifyLoneGenesis ( stream ) ; } ) ; await this . handlePinOpts ( state$ , opts ) ; return state$ ; } param commitId param opts async loadAtCommit ( commitId , opts ) { const base$ = await this . load ( commitId . baseID , opts ) ; return this . stateManager . atCommit ( base$ , commitId ) ; } param streamId param opts async loadAtTime ( streamId , opts ) { const base$ = await this . load ( streamId . baseID , opts ) ; return this . stateManager . atTime ( base$ , opts . atTime ) ; } param streamId param commit param opts async applyCommit ( streamId , commit , opts ) { const state$ = await this . stateManager . applyCommit ( streamId , commit , opts ) ; await this . applyWriteOpts ( state$ , opts ) ; return state$ ; } param state$ param opts private async applyWriteOpts ( state$ , opts ) { this . stateManager . applyWriteOpts ( state$ , opts ) ; await this . handlePinOpts ( state$ , opts ) ; } async handlePinOpts ( state$ , opts ) { if ( opts . pin ) { await this . pin ( state$ ) ; } else if ( opts . pin === false ) { await this . unpin ( state$ ) ; } } param streamId param opts async applyCreateOpts ( streamId , opts ) { const state = await this . load ( streamId , opts ) ; await this . applyWriteOpts ( state , opts ) ; return state ; } /**
     * Return a stream, either from cache or re-constructed from state store, but will not load from the network.
     * Adds the stream to cache.
     */ async get ( streamId ) { return this . loadingQ . forStream ( streamId ) . run ( async ( ) => { const fromMemory = this . fromMemory ( streamId ) ; if ( fromMemory ) return fromMemory ; return this . fromStateStore ( streamId ) ; } ) ; } /**
     * Return a stream state, either from cache or from state store.
     */ async streamState ( streamId ) { const fromMemory = this . inmemory . get ( streamId . toString ( ) ) ; if ( fromMemory ) { return fromMemory . state ; } else { return this . #deps . pinStore . stateStore . load ( streamId ) ; } } /**
     * Adds the stream's RunningState to the in-memory cache and subscribes the Repository's global feed$ to receive changes emitted by that RunningState
     */ add ( state$ ) { this . inmemory . set ( state$ . id . toString ( ) , state$ ) ; } pin ( state$ , force ? ) { return this . #deps . pinStore . add ( state$ , force ) ; } async unpin ( state$ , opts ? ) { if ( opts ?. publish ) { this . stateManager . publishTip ( state$ ) ; } return this . #deps . pinStore . rm ( state$ ) ; } /**
     * List pinned streams as array of StreamID strings.
     * If `streamId` is passed, indicate if it is pinned.
     */ async listPinned ( streamId ? ) { return this . #deps . pinStore . ls ( streamId ) ; } param init updates$ ( init ) { return new Observable < StreamState > ( ( subscriber ) => { const id = new StreamID ( init . type , init . log [ 0 ] . cid ) ; this . get ( id ) . then ( ( found ) => { const state$ = found || new RunningState ( init , false ) ; this . inmemory . endure ( id . toString ( ) , state$ ) ; state$ . subscribe ( subscriber ) . add ( ( ) => { if ( state$ . observers . length === 0 ) { this . inmemory . free ( id . toString ( ) ) ; } } ) ; } ) ; } ) ; } async close ( ) { await this . loadingQ . close ( ) ; await this . executionQ . close ( ) ; Array . from ( this . inmemory ) . forEach ( ( [ id , stream ] ) => { this . inmemory . delete ( id ) ; stream . complete ( ) ; } ) ; await this . #deps . pinStore . close ( ) ; } }