Unable to resolve handler when using BusPostgresModule

[jabstone]

Hello. I have been using the in-memory persistence provider with no issues, works great. I recently tried to use the postgres persistence provider and I'm getting an error that I haven't been able to figure out. Perhaps I'm doing something wrong. When I create a new message to start a workflow using the postgres provider, I'm getting the following error:

{"@message":"Message read from transport [object Object]","@timestamp":"2020-12-10T05:08:04.230Z","@fields":{"name":"ServiceBus","level":"debug"}}
{"@message":"Resolving handlers for message.","@timestamp":"2020-12-10T05:08:04.232Z","@fields":{"name":"HandlerRegistry","receivedMessage":{"orderId":"984c243e-28eb-42b7-906f-cefe973bb2ef","$name":"tc/pipeline/ortho-order","$version":1},"level":"debug"}}
{"@message":"Could not resolve handler from the IoC container.","@timestamp":"2020-12-10T05:08:04.233Z","@fields":{"name":"HandlerRegistry","receivedMessage":{"orderId":"984c243e-28eb-42b7-906f-cefe973bb2ef","$name":"tc/pipeline/ortho-order","$version":1},"error":{"name":"Error","message":"No matching bindings found for serviceIdentifier: Symbol(node-ts/bus/workflow/OrderWorkflow-tc/pipeline/ortho-order-started-by-proxy)","stack":"Error: No matching bindings found for serviceIdentifier: Symbol(node-ts/bus/workflow/OrderWorkflow-tc/pipeline/ortho-order-started-by-proxy)\n    at _validateActiveBindingCount (/app/node_modules/inversify/lib/planning/planner.js:62:23)\n    at _getActiveBindings (/app/node_modules/inversify/lib/planning/planner.js:48:5)\n    at _createSubRequests (/app/node_modules/inversify/lib/planning/planner.js:85:26)\n    at Object.plan (/app/node_modules/inversify/lib/planning/planner.js:136:9)\n    at /app/node_modules/inversify/lib/container/container.js:317:37\n    at Container._get (/app/node_modules/inversify/lib/container/container.js:310:44)\n    at Container.get (/app/node_modules/inversify/lib/container/container.js:230:21)\n    at Object.resolveHandler (/app/node_modules/@node-ts/bus-core/dist/handler/handler-registry.js:66:38)\n    at dispatchMessageToHandler (/app/node_modules/@node-ts/bus-core/dist/service-bus/service-bus.js:155:41)\n    at /app/node_modules/@node-ts/bus-core/dist/service-bus/service-bus.js:116:58\n    at Array.map (<anonymous>)\n    at ServiceBus.dispatchMessageToHandlers (/app/node_modules/@node-ts/bus-core/dist/service-bus/service-bus.js:116:43)\n    at ServiceBus.handleNextMessage (/app/node_modules/@node-ts/bus-core/dist/service-bus/service-bus.js:92:32)\n    at async ServiceBus.applicationLoop (/app/node_modules/@node-ts/bus-core/dist/service-bus/service-bus.js:81:13)"},"level":"error"}}

I believe I have everything hooked up correctly. The logs show the postgres provider initializes correctly and I have confirmed the schema and tables exist. However, when an I send an event to start the workflow, I get the error above. Here are the logs showing the postgres provider is initializing successfully.

{"@message":"Initializing postgres persistence...","@timestamp":"2020-12-10T05:07:27.872Z","@fields":{"name":"PostgresPersistence","level":"info"}}
{"@message":"Postgres persistence initialized","@timestamp":"2020-12-10T05:07:27.893Z","@fields":{"name":"PostgresPersistence","level":"info"}}
{"@message":"Initializing workflow","@timestamp":"2020-12-10T05:07:27.894Z","@fields":{"name":"PostgresPersistence","workflowData":"tc/pipeline/ortho-order-data","level":"info"}}
{"@message":"Ensuring postgres table for workflow data exists","@timestamp":"2020-12-10T05:07:27.894Z","@fields":{"name":"PostgresPersistence","sql":"\n      create table if not exists \"workflows\".\"tcpipelineortho-order-data\" (\n        id uuid not null primary key,\n        version integer not null,\n        data jsonb not null\n      );\n    ","level":"debug"}}
{"@message":"Ensuring primary index exists","@timestamp":"2020-12-10T05:07:27.906Z","@fields":{"name":"PostgresPersistence","createPrimaryIndexSql":"\n      DO\n      $$\n      BEGIN\n        IF to_regclass('workflows.\"workflows_tcpipelineortho-order-data_id_version_idx\"') IS NULL THEN\n          CREATE INDEX \"workflows_tcpipelineortho-order-data_id_version_idx\" ON \"workflows\".\"tcpipelineortho-order-data\" (id, version);\n        END IF;\n      END\n      $$;\n    ","level":"debug"}}
{"@message":"Ensuring secondary index exists","@timestamp":"2020-12-10T05:07:27.906Z","@fields":{"name":"PostgresPersistence","createSecondaryIndex":"\n        DO\n        $$\n        BEGIN\n          IF to_regclass('workflows.\"workflows_tcpipelineortho-order-data_orderId_idx\"') IS NULL THEN\n            CREATE INDEX\n              \"workflows_tcpipelineortho-order-data_orderId_idx\"\n            ON\n              \"workflows\".\"tcpipelineortho-order-data\" ((data->>'orderId'))\n            WHERE\n              (data->>'orderId') is not null;\n          END IF;\n        END\n        $$;\n      ","level":"debug"}}

The exact same test using the in-memory provider works fine as indicated in the logs below.

{"@message":"Worker started","@timestamp":"2020-12-10T05:11:10.963Z","@fields":{"name":"ServiceBus","runningParallelWorkerCount":1,"level":"debug"}}
{"@message":"send","@timestamp":"2020-12-10T05:11:38.090Z","@fields":{"name":"ServiceBus","command":{"orderId":"984c243e-28eb-42b7-906f-cefe973bb2ef","$name":"tc/pipeline/ortho-order","$version":1},"level":"debug"}}
{"@message":"Message read from transport [object Object]","@timestamp":"2020-12-10T05:11:38.139Z","@fields":{"name":"ServiceBus","level":"debug"}}
{"@message":"Resolving handlers for message.","@timestamp":"2020-12-10T05:11:38.139Z","@fields":{"name":"HandlerRegistry","receivedMessage":{"orderId":"984c243e-28eb-42b7-906f-cefe973bb2ef","$name":"tc/pipeline/ortho-order","$version":1},"level":"debug"}}
{"@message":"Getting workflow data for message [object Object]","@timestamp":"2020-12-10T05:11:38.140Z","@fields":{"name":"StartedByProxy","messageOptions":{"attributes":{},"stickyAttributes":{}},"level":"debug"}}
{"@message":"Workflow data retrieved [object Object]","@timestamp":"2020-12-10T05:11:38.141Z","@fields":{"name":"StartedByProxy","workflowData":[{"$version":0,"$name":"tc/pipeline/ortho-order-data","$status":"running","$workflowId":"f200b85d-d75c-41d1-a06e-6d53d5a910e3"}],"level":"debug"}}
{"@message":"Received new order event. Requesting stitching for orderId: 984c243e-28eb-42b7-906f-cefe973bb2ef","@timestamp":"2020-12-10T05:11:38.141Z","@fields":{"name":"OrderWorkflow","level":"info"}}
{"@message":"send","@timestamp":"2020-12-10T05:11:38.141Z","@fields":{"name":"ServiceBus","command":{"orderId":"984c243e-28eb-42b7-906f-cefe973bb2ef","$name":"tc/pipeline/stitching","$version":1},"level":"debug"}}
{"@message":"Changes detected in workflow data and will be persisted.","@timestamp":"2020-12-10T05:11:38.141Z","@fields":{"name":"StartedByProxy","workflowId":"f200b85d-d75c-41d1-a06e-6d53d5a910e3","workflowName":"OrderWorkflowData","level":"debug"}}
{"@message":"Saving workflow data","@timestamp":"2020-12-10T05:11:38.142Z","@fields":{"name":"StartedByProxy","data":{"$version":0,"$name":"tc/pipeline/ortho-order-data","$status":"running","$workflowId":"f200b85d-d75c-41d1-a06e-6d53d5a910e3","orderId":"984c243e-28eb-42b7-906f-cefe973bb2ef","status":1},"level":"info"}}
{"@message":"Message dispatched to all handlers [object Object]","@timestamp":"2020-12-10T05:11:38.142Z","@fields":{"name":"ServiceBus","level":"debug"}}

I'm posting this message here in case this is a bug when trying to use the postgres provider.

Here is my container that loads my modules.

import { Container } from 'inversify'
import { BusModule } from '@node-ts/bus-core'
import { BusWorkflowModule } from '@node-ts/bus-workflow'
import { LoggerModule } from '@node-ts/logger-core'
import { WinstonModule } from '@node-ts/logger-winston'
import { BusRabbitMqModule } from '@node-ts/bus-rabbitmq'
import { PipelineContainerModule } from './PipelineContainerModule'
 import { BusPostgresModule } from '@node-ts/bus-postgres'

export class WorkflowContainer extends Container {

    constructor() {
        super();
        this.load(
            new LoggerModule(),
            new WinstonModule(),
            new BusModule(),
            new BusWorkflowModule(),
            new BusRabbitMqModule(),      
            new BusPostgresModule(),
            new PipelineContainerModule()                
        );
    }
}

Here is a chunk of my index.ts that wires everything together.

       const workflowContainer = new WorkflowContainer();
        workflowContainer.rebind(WINSTON_SYMBOLS.WinstonConfiguration).to(LoggerConfiguration)

        //Bind MessageQueue to container
        const rabbitConfiguration: RabbitMqTransportConfiguration = {
            queueName: 'accounts-application-queue',
            connectionString: 'amqp:https://xxx:[email protected]',
            maxRetries: 5
          }
        workflowContainer.bind(BUS_RABBITMQ_SYMBOLS.TransportConfiguration).toConstantValue(rabbitConfiguration);

        //Bind Postgres to container
        const pgConfiguration: PostgresConfiguration = {
            connection: {
              connectionString: 'postgres:https://postgres:[email protected]:5432/postgres'
            },
            schemaName: 'workflows'
          }
          workflowContainer.bind(BUS_POSTGRES_SYMBOLS.PostgresConfiguration).toConstantValue(pgConfiguration);


        // Register the workflow with the registry so all the underlying messaging infrastructure can be initialized
        const workflowRegistry = workflowContainer.get<WorkflowRegistry>(BUS_WORKFLOW_SYMBOLS.WorkflowRegistry)
        workflowRegistry.register(OrderWorkflow, OrderWorkflowData)
        workflowRegistry.initializeWorkflows();

        // Register event registration handlers
        const bootstrap = workflowContainer.get<ApplicationBootstrap>(BUS_SYMBOLS.ApplicationBootstrap)
        bootstrap.registerHandler(TestHandler);
        await bootstrap.initialize(workflowContainer);        

Any insight would be greatly appreciated and thanks for making this excellent project.

[adenhertog]

hey @jabstone

can you try again after changing workflowRegistry.initializeWorkflows(); to await workflowRegistry.initializeWorkflows(); and let me know how it goes?

[jabstone]

Oh wow, sometimes it's the little things. I'm sorry I missed that one but thank you so much for finding my error. That was my problem. Really appreciate it!