import express , { Request , Response } from 'express' ; import type { CeramicConfig } from '@ceramicnetwork/core' ; import Ceramic from '@ceramicnetwork/core' ; import { RotatingFileStream } from '@ceramicnetwork/logger' ; import { buildIpfsConnection } from './build-ipfs-connection.util' ; import { S3StateStore } from './s3-state-store' ; import { DiagnosticsLogger , LoggerProvider , MultiQuery , StreamUtils , SyncOptions , } from '@ceramicnetwork/common' ; import StreamID , { StreamType } from '@ceramicnetwork/streamid' ; import ThreeIdResolver from '@ceramicnetwork/3id-did-resolver' ; import KeyDidResolver from 'key-did-resolver' ; import PkhDidResolver from 'pkh-did-resolver' ; import EthrDidResolver from 'ethr-did-resolver' ; import NftDidResolver from 'nft-did-resolver' ; import SafeDidResolver from 'safe-did-resolver' ; import { DID } from 'dids' ; import cors from 'cors' ; import { errorHandler } from './daemon/error-handler' ; import { addAsync , ExpressWithAsync , Router } from '@awaitjs/express' ; import { logRequests } from './daemon/log-requests' ; import type { Server } from 'http' ; import { DaemonConfig , StateStoreMode } from './daemon-config' ; import type { ResolverRegistry } from 'did-resolver' ; const packageJson = require ( '../package.json' ) ; const DEFAULT_HOSTNAME = '0.0.0.0' ; const DEFAULT_PORT = 7007 ; const HEALTHCHECK_RETRIES = 3 ; interface MultiQueryWithDocId extends MultiQuery { docId ? : string ; } interface MultiQueries { queries : Array < MultiQueryWithDocId > ; timeout ? : number ; } const SYNC_OPTIONS_MAP = { 'prefer-cache' : SyncOptions . PREFER_CACHE , 'sync-always' : SyncOptions . SYNC_ALWAYS , 'never-sync' : SyncOptions . NEVER_SYNC , } ; export function makeCeramicConfig ( opts ) { const loggerProvider = new LoggerProvider ( opts . logger , ( logPath ) => { return new RotatingFileStream ( logPath , true ) ; } ) ; const ceramicConfig = { loggerProvider , gateway : opts . node . gateway || false , anchorServiceUrl : opts . anchor . anchorServiceUrl , ethereumRpcUrl : opts . anchor . ethereumRpcUrl , ipfsPinningEndpoints : opts . ipfs . pinningEndpoints , networkName : opts . network . name , pubsubTopic : opts . network . pubsubTopic , validateStreams : opts . node . validateStreams , syncOverride : SYNC_OPTIONS_MAP [ opts . node . syncOverride ] , } ; if ( opts . stateStore ?. mode == StateStoreMode . FS ) { ceramicConfig . stateStoreDirectory = opts . stateStore . localDirectory ; } return ceramicConfig ; } param opts function parseQueryObject ( opts ) { const typedOpts = { } ; for ( const [ key , value ] of Object . entries ( opts ) ) { if ( typeof value == 'string' ) { if ( value === 'true' ) { typedOpts [ key ] = true ; } else if ( value === 'false' ) { typedOpts [ key ] = false ; } else if ( ! isNaN ( parseInt ( value ) ) ) { typedOpts [ key ] = parseInt ( value ) ; } else { typedOpts [ key ] = value ; } } else { typedOpts [ key ] = value ; } } return typedOpts ; } param opts function upconvertLegacySyncOption ( opts ) { if ( typeof opts ?. sync == 'boolean' ) { if ( opts . sync ) { opts . sync = SyncOptions . SYNC_ALWAYS ; } else { opts . sync = SyncOptions . PREFER_CACHE ; } } } /**
 * Prepare DID resolvers to use in the daemon.
 */ function makeResolvers ( ceramic , ceramicConfig , opts ) { let result = { ... KeyDidResolver . getResolver ( ) , ... PkhDidResolver . getResolver ( ) , ... ThreeIdResolver . getResolver ( ceramic ) , ... NftDidResolver . getResolver ( { ceramic : ceramic , ... opts . didResolvers ?. nftDidResolver , } ) , ... SafeDidResolver . getResolver ( { ceramic : ceramic , ... opts . didResolvers ?. safeDidResolver , } ) , } ; if ( opts . didResolvers ?. ethrDidResolver ?. networks && opts . didResolvers ?. ethrDidResolver ?. networks . length > 0 ) { result = { ... result , ... EthrDidResolver . getResolver ( opts . didResolvers . ethrDidResolver ) } ; } else if ( ceramicConfig . ethereumRpcUrl ) { result = { ... result , ... EthrDidResolver . getResolver ( { networks : [ { rpcUrl : ceramicConfig . ethereumRpcUrl , } , ] , } ) , } ; } return result ; } /**
 * Ceramic daemon implementation
 */ export class CeramicDaemon { private server ? : Server ; private readonly app : ExpressWithAsync ; private readonly diagnosticsLogger : DiagnosticsLogger ; public hostname : string ; public port : number ; constructor ( public ceramic , private readonly opts ) { this . diagnosticsLogger = ceramic . loggerProvider . getDiagnosticsLogger ( ) ; this . port = this . opts . httpApi ?. port || DEFAULT_PORT ; this . hostname = this . opts . httpApi ?. hostname || DEFAULT_HOSTNAME ; this . app = addAsync ( express ( ) ) ; this . app . set ( 'trust proxy' , true ) ; this . app . use ( express . json ( { limit : '1mb' } ) ) ; this . app . use ( cors ( { origin : opts . httpApi ?. corsAllowedOrigins , maxAge : 7200 , } ) ) ; this . app . use ( logRequests ( ceramic . loggerProvider ) ) ; this . registerAPIPaths ( this . app , opts . node ?. gateway ) ; this . app . use ( errorHandler ( this . diagnosticsLogger ) ) ; } async listen ( ) { return new Promise < void > ( ( resolve ) => { this . server = this . app . listen ( this . port , this . hostname , ( ) => { this . diagnosticsLogger . imp ( `Ceramic API running on ${ this . hostname }:${ this . port }'` ) ; resolve ( ) ; } ) ; this . server . keepAliveTimeout = 60 * 1000 ; } ) ; } param opts static async create ( opts ) { const ceramicConfig = makeCeramicConfig ( opts ) ; const ipfs = await buildIpfsConnection ( opts . ipfs . mode , opts . network ?. name , ceramicConfig . loggerProvider . getDiagnosticsLogger ( ) , opts . ipfs ?. host ) ; const [ modules , params ] = Ceramic . _processConfig ( ipfs , ceramicConfig ) ; modules . loggerProvider . getDiagnosticsLogger ( ) . imp ( `Starting Ceramic Daemon at version ${ packageJson . version } with config: \n${ JSON . stringify ( opts , null , 2 ) }` ) ; if ( opts . stateStore ?. mode == StateStoreMode . S3 ) { const s3StateStore = new S3StateStore ( opts . stateStore ?. s3Bucket ) ; modules . pinStoreFactory . setStateStore ( s3StateStore ) ; } const ceramic = new Ceramic ( modules , params ) ; await ceramic . _init ( true ) ; const did = new DID ( { resolver : makeResolvers ( ceramic , ceramicConfig , opts ) } ) ; await ceramic . setDID ( did ) ; const daemon = new CeramicDaemon ( ceramic , opts ) ; await daemon . listen ( ) ; return daemon ; } registerAPIPaths ( app , gateway ) { const baseRouter = Router ( ) ; const commitsRouter = Router ( ) ; const documentsRouter = Router ( ) ; const multiqueriesRouter = Router ( ) ; const nodeRouter = Router ( ) ; const pinsRouter = Router ( ) ; const recordsRouter = Router ( ) ; const streamsRouter = Router ( ) ; app . use ( '/api/v0' , baseRouter ) ; baseRouter . use ( '/commits' , commitsRouter ) ; baseRouter . use ( '/documents' , documentsRouter ) ; baseRouter . use ( '/multiqueries' , multiqueriesRouter ) ; baseRouter . use ( '/node' , nodeRouter ) ; baseRouter . use ( '/pins' , pinsRouter ) ; baseRouter . use ( '/records' , recordsRouter ) ; baseRouter . use ( '/streams' , streamsRouter ) ; baseRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; commitsRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; documentsRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; multiqueriesRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; nodeRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; pinsRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; recordsRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; streamsRouter . use ( errorHandler ( this . diagnosticsLogger ) ) ; commitsRouter . getAsync ( '/:streamid' , this . commits . bind ( this ) ) ; multiqueriesRouter . postAsync ( '/' , this . multiQuery . bind ( this ) ) ; streamsRouter . getAsync ( '/:streamid' , this . state . bind ( this ) ) ; streamsRouter . getAsync ( '/:streamid/content' , this . content . bind ( this ) ) ; pinsRouter . getAsync ( '/:streamid' , this . listPinned . bind ( this ) ) ; pinsRouter . getAsync ( '/' , this . listPinned . bind ( this ) ) ; nodeRouter . getAsync ( '/chains' , this . getSupportedChains . bind ( this ) ) ; nodeRouter . getAsync ( '/healthcheck' , this . healthcheck . bind ( this ) ) ; documentsRouter . getAsync ( '/:docid' , this . stateOld . bind ( this ) ) ; recordsRouter . getAsync ( '/:streamid' , this . commits . bind ( this ) ) ; if ( ! gateway ) { streamsRouter . postAsync ( '/' , this . createStreamFromGenesis . bind ( this ) ) ; streamsRouter . postAsync ( '/:streamid/anchor' , this . requestAnchor . bind ( this ) ) ; commitsRouter . postAsync ( '/' , this . applyCommit . bind ( this ) ) ; pinsRouter . postAsync ( '/:streamid' , this . pinStream . bind ( this ) ) ; pinsRouter . deleteAsync ( '/:streamid' , this . unpinStream . bind ( this ) ) ; documentsRouter . postAsync ( '/' , this . createDocFromGenesis . bind ( this ) ) ; recordsRouter . postAsync ( '/' , this . applyCommit . bind ( this ) ) ; } else { streamsRouter . postAsync ( '/' , this . createReadOnlyStreamFromGenesis . bind ( this ) ) ; commitsRouter . postAsync ( '/' , this . _notSupported . bind ( this ) ) ; pinsRouter . postAsync ( '/:streamid' , this . _notSupported . bind ( this ) ) ; pinsRouter . deleteAsync ( '/:streamid' , this . _notSupported . bind ( this ) ) ; documentsRouter . postAsync ( '/' , this . createReadOnlyDocFromGenesis . bind ( this ) ) ; recordsRouter . postAsync ( '/' , this . _notSupported . bind ( this ) ) ; } } dev async healthcheck ( req , res ) { for ( let i = 0 ; i < HEALTHCHECK_RETRIES ; i ++ ) { try { if ( await this . ceramic . ipfs . isOnline ( ) ) { res . status ( 200 ) . send ( 'Alive!' ) ; return ; } } catch ( e ) { this . diagnosticsLogger . err ( `Error checking IPFS status: ${ e }` ) ; } } res . status ( 503 ) . send ( 'IPFS unreachable' ) ; } dev deprecated async createDocFromGenesis ( req , res ) { const { doctype , genesis , docOpts } = req . body ; upconvertLegacySyncOption ( docOpts ) ; const type = StreamType . codeByName ( doctype ) ; const doc = await this . ceramic . createStreamFromGenesis ( type , StreamUtils . deserializeCommit ( genesis ) , docOpts ) ; res . json ( { streamId : doc . id . toString ( ) , docId : doc . id . toString ( ) , state : StreamUtils . serializeState ( doc . state ) , } ) ; } dev async createStreamFromGenesis ( req , res ) { const { type , genesis , opts } = req . body ; const stream = await this . ceramic . createStreamFromGenesis ( type , StreamUtils . deserializeCommit ( genesis ) , opts ) ; res . json ( { streamId : stream . id . toString ( ) , state : StreamUtils . serializeState ( stream . state ) } ) ; } dev async requestAnchor ( req , res ) { const streamId = StreamID . fromString ( req . params . streamid ) ; const opts = req . body . opts ; const anchorStatus = await this . ceramic . requestAnchor ( streamId , opts ) ; res . json ( { streamId : streamId . toString ( ) , anchorStatus } ) ; } dev TODO deprecated async createReadOnlyDocFromGenesis ( req , res ) { const { doctype , genesis , docOpts } = req . body ; upconvertLegacySyncOption ( docOpts ) ; const type = StreamType . codeByName ( doctype ) ; const readOnlyDocOpts = { ... docOpts , anchor : false , publish : false } ; const doc = await this . ceramic . createStreamFromGenesis ( type , StreamUtils . deserializeCommit ( genesis ) , readOnlyDocOpts ) ; res . json ( { streamId : doc . id . toString ( ) , docId : doc . id . toString ( ) , state : StreamUtils . serializeState ( doc . state ) , } ) ; } dev TODO async createReadOnlyStreamFromGenesis ( req , res ) { const { type , genesis , opts } = req . body ; const readOnlyOpts = { ... opts , anchor : false , publish : false } ; const stream = await this . ceramic . createStreamFromGenesis ( type , StreamUtils . deserializeCommit ( genesis ) , readOnlyOpts ) ; res . json ( { streamId : stream . id . toString ( ) , state : StreamUtils . serializeState ( stream . state ) } ) ; } /**
     * Get stream state
     */ async state ( req , res ) { const opts = parseQueryObject ( req . query ) ; const stream = await this . ceramic . loadStream ( req . params . streamid , opts ) ; res . json ( { streamId : stream . id . toString ( ) , state : StreamUtils . serializeState ( stream . state ) } ) ; } deprecated async stateOld ( req , res ) { const opts = parseQueryObject ( req . query ) ; upconvertLegacySyncOption ( opts ) ; const doc = await this . ceramic . loadStream ( req . params . docid , opts ) ; res . json ( { docId : doc . id . toString ( ) , state : StreamUtils . serializeState ( doc . state ) } ) ; } /**
     * Get all document commits
     */ async commits ( req , res ) { const streamId = StreamID . fromString ( req . params . streamid || req . params . docid ) ; const commits = await this . ceramic . loadStreamCommits ( streamId ) ; const serializedCommits = commits . map ( ( r ) => { return { cid : r . cid , value : StreamUtils . serializeCommit ( r . value ) , } ; } ) ; res . json ( { streamId : streamId . toString ( ) , docId : streamId . toString ( ) , commits : serializedCommits , } ) ; } /**
     * Apply one commit to the existing document
     */ async applyCommit ( req , res ) { const { docId , commit , docOpts } = req . body ; const opts = req . body . opts || docOpts ; upconvertLegacySyncOption ( opts ) ; opts . throwOnInvalidCommit = opts . throwOnInvalidCommit ?? true ; const streamId = req . body . streamId || docId ; if ( ! ( streamId && commit ) ) { throw new Error ( 'streamId and commit are required in order to apply commit' ) ; } const stream = await this . ceramic . applyCommit ( streamId , StreamUtils . deserializeCommit ( commit ) , opts ) ; res . json ( { streamId : stream . id . toString ( ) , docId : stream . id . toString ( ) , state : StreamUtils . serializeState ( stream . state ) , } ) ; } /**
     * Load multiple documents and paths using an array of multiqueries
     */ async multiQuery ( req , res ) { let { queries } = < MultiQueries > req . body ; const { timeout } = < MultiQueries > req . body ; queries = queries . map ( ( q ) : MultiQuery => { if ( q . docId ) { q . streamId = q . docId ; delete q . docId ; } return q ; } ) ; const results = await this . ceramic . multiQuery ( queries , timeout ) ; const response = Object . entries ( results ) . reduce ( ( acc , e ) => { const [ k , v ] = e ; acc [ k ] = StreamUtils . serializeState ( v . state ) ; return acc ; } , { } ) ; res . json ( response ) ; } /**
     * Render the most recent version of a stream's contents
     */ async content ( req , res ) { const opts = parseQueryObject ( req . query ) ; const stream = await this . ceramic . loadStream ( req . params . streamid , opts ) ; res . json ( stream . content ) ; } /**
     * Pin stream
     */ async pinStream ( req , res ) { const streamId = StreamID . fromString ( req . params . streamid || req . params . docid ) ; const { force } = req . body ; await this . ceramic . pin . add ( streamId , force ) ; res . json ( { streamId : streamId . toString ( ) , docId : streamId . toString ( ) , isPinned : true , } ) ; } /**
     * Unpin stream
     */ async unpinStream ( req , res ) { const streamId = StreamID . fromString ( req . params . streamid || req . params . docid ) ; const { opts } = req . body ; await this . ceramic . pin . rm ( streamId , opts ) ; res . json ( { streamId : streamId . toString ( ) , docId : streamId . toString ( ) , isPinned : false , } ) ; } /**
     * List pinned streams
     */ async listPinned ( req , res ) { let streamId ; if ( req . params . streamid || req . params . docid ) { streamId = StreamID . fromString ( req . params . streamid || req . params . docid ) ; } const pinnedStreamIds = [ ] ; const iterator = await this . ceramic . pin . ls ( streamId ) ; for await ( const id of iterator ) { pinnedStreamIds . push ( id ) ; } res . json ( { pinnedStreamIds , pinnedDocIds : pinnedStreamIds } ) ; } async _notSupported ( req , res ) { res . status ( 400 ) . json ( { error : 'Method not supported by read only Ceramic Gateway' } ) ; } async getSupportedChains ( req , res ) { const supportedChains = await this . ceramic . getSupportedChains ( ) ; res . json ( { supportedChains } ) ; } /**
     * Close Ceramic daemon
     */ async close ( ) { return new Promise < void > ( ( resolve , reject ) => { if ( ! this . server ) resolve ( ) ; this . server . close ( ( err ) => { if ( err ) { reject ( err ) ; } else { resolve ( ) ; } } ) ; } ) ; } }