event

type BatchScannedConsumer ¶

type BatchScannedConsumer struct {
	// contains filtered or unexported fields
}

BatchScannedConsumer moves inventory in reaction to a scan: it credits what the batch produced, releases the reservations covering units that seconds and waste mean will never exist, and draws down the materials the step consumed.

It is one subscriber to core.event.batch_scanned among however many the account eventually wants; it owns no part of the event beyond deciding what inventory should do about it.

The whole reaction commits as one transaction. The delivery is retried with backoff and replayed from the top on failure, and none of these writes is individually idempotent — a step that failed after allocating drew the same receipts again on the retry, consuming stock that never moved.

func NewBatchScannedConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	repos domain.RepoFactory,
	txManager db.TransactionManager[*sqlc.Queries, domain.RepoFactory],
) *BatchScannedConsumer

func (c *BatchScannedConsumer) Listen(ctx context.Context) error

func (c *BatchScannedConsumer) ReplayMessage(ctx context.Context, msg amqp.Delivery) error

type BulkOperationConsumer ¶

type BulkOperationConsumer struct {
	// contains filtered or unexported fields
}

BulkOperationConsumer is the single consumer behind every async bulk operation. Each bulk endpoint enqueues a {job_id} command; this consumer loads the job (whose resolved, validated payload was stored at enqueue time), restores the originating identity, and hands off to the entity's executor. The only variance between operations is data — the queue, the inbox handler key, and the executor — so there is one implementation instead of one near-identical file per operation. The message inbox de-dupes redeliveries so retries converge on one visible outcome.

func NewBulkOperationConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	op messaging.BulkOperation,
	execute BulkOperationExecutor,
) *BulkOperationConsumer

func (c *BulkOperationConsumer) Listen(ctx context.Context) error

type BulkOperationExecutor ¶

type BulkOperationExecutor func(context.Context, domain.BulkOperationJobEvent) *apierror.APIError

BulkOperationExecutor performs the write phase of an enqueued bulk operation: it loads the job named by the event and runs its resolved rows. Each entity's service supplies one — its Execute method value (e.g. unitGroupSvc.ExecuteBulkUpsertUnitGroups).

type CustomerRegisteredPublisher ¶

type CustomerRegisteredPublisher struct{}

CustomerRegisteredPublisher writes customer-registered events to the outbox so the notification-service can notify the seller's customer-service group out-of-band from the registration response.

func NewCustomerRegisteredPublisher() *CustomerRegisteredPublisher

func (p *CustomerRegisteredPublisher) Publish(ctx context.Context, outboxRepo messaging.OutboxRepo, data messaging.CustomerRegisteredData) *apierror.APIError

type ExecuteProductionStepConsumer ¶

type ExecuteProductionStepConsumer struct {
	// contains filtered or unexported fields
}

ExecuteProductionStepConsumer processes execute-production-step commands. It calculates inventory changes from batch mutations (initialize, move, merge, split) and creates inventory receipts/issues accordingly.

func NewExecuteProductionStepConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	queries *sqlc.Queries,
	repos domain.RepoFactory,
	txManager db.TransactionManager[*sqlc.Queries, domain.RepoFactory],
) *ExecuteProductionStepConsumer

func (c *ExecuteProductionStepConsumer) Listen(ctx context.Context) error

type ExportConsumer ¶

type ExportConsumer struct {
	// contains filtered or unexported fields
}

consumes one export command. An export carries the same {job_id} payload a bulk operation does, so the shape mirrors BulkOperationConsumer; only the work differs.

func NewExportConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	op messaging.ExportOperation,
	render ExportRenderer,
) *ExportConsumer

func (c *ExportConsumer) Listen(ctx context.Context) error

type ExportRenderer ¶

type ExportRenderer func(context.Context, domain.BulkOperationJobEvent) *apierror.APIError

renders an accepted export and settles the job tracking it. Each entity's service supplies one — its BuildExport method value.

type GenerateProductionScheduleConsumer ¶

type GenerateProductionScheduleConsumer struct {
	// contains filtered or unexported fields
}

GenerateProductionScheduleConsumer runs the solve the generation cadence queued.

The solve lives here rather than in the cadence tick because it takes minutes on a real tenant: running it inside the scheduler lease would block every other account behind whichever one is currently solving.

func NewGenerateProductionScheduleConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	scheduleSvc domain.ProductionScheduleSvc,
	repos domain.RepoFactory,
) *GenerateProductionScheduleConsumer

func (c *GenerateProductionScheduleConsumer) Listen(ctx context.Context) error

type HubspotSyncConsumer ¶

type HubspotSyncConsumer struct {
	// contains filtered or unexported fields
}

HubspotSyncConsumer runs the long-running HubSpot backfill passes (preview and execute) out-of-band. The queue is bound to both command routing keys; handleMessage dispatches on the routing key.

func NewHubspotSyncConsumer(rabbitmq messaging.MessageBroker, inboxRepo messaging.InboxRepo, hubspotSync hubspotsync.Service) *HubspotSyncConsumer

func (c *HubspotSyncConsumer) Listen(ctx context.Context) error

type InventoryReceivedConsumer ¶

type InventoryReceivedConsumer struct {
	// contains filtered or unexported fields
}

InventoryReceivedConsumer offers newly available stock to the demand that went short waiting for it.

An issue goes open when it was asked for more than the shelf could cover. It stays short until stock arrives, and this is what notices that it has. That used to be a nightly sweep over every open issue in the account, which meant an order could sit unfilled for most of a day with the stock to fill it already on the floor — and made a scan that failed to allocate indistinguishable from one that had nothing to allocate, because the sweep quietly fixed both.

func NewInventoryReceivedConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	repos domain.RepoFactory,
	txManager db.TransactionManager[*sqlc.Queries, domain.RepoFactory],
) *InventoryReceivedConsumer

func (c *InventoryReceivedConsumer) Listen(ctx context.Context) error

type ItemCostBasisChangedConsumer ¶

type ItemCostBasisChangedConsumer struct {
	// contains filtered or unexported fields
}

ItemCostBasisChangedConsumer recomputes the cost of everything downstream of a change.

A cost is only as current as the inputs it was last calculated from, so moving a material's price or editing a production step leaves every part and product built on it stale. This walks the production graph outwards from the change and recalculates what it reaches.

It replaces a nightly job that recomputed every item with a production, for one hardcoded account, whether or not anything had changed — which meant a price change entered on Tuesday was wrong on every quote until Wednesday morning, and right by accident rather than by design.

func NewItemCostBasisChangedConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	repos domain.RepoFactory,
	itemSvc domain.ItemSvc,
) *ItemCostBasisChangedConsumer

func (c *ItemCostBasisChangedConsumer) Listen(ctx context.Context) error

type PurgeAccountDataPayload ¶

type PurgeAccountDataPayload struct {
	AccountID string `json:"account_id"`
}

PurgeAccountDataPayload is the payload for CoreCmdPurgeAccountData messages.

type PurgeConsumer ¶

type PurgeConsumer struct {
	// contains filtered or unexported fields
}

func NewPurgeConsumer(rabbitmq messaging.MessageBroker, inboxRepo messaging.InboxRepo, purgeRepo *repository.PurgeRepo) *PurgeConsumer

func (c *PurgeConsumer) Listen(ctx context.Context) error

type RecalcItemBurnRateConsumer ¶ added in v1.1.8

type RecalcItemBurnRateConsumer struct {
	// contains filtered or unexported fields
}

RecalcItemBurnRateConsumer recomputes an item's burn rate from its recent consumption history, off the transaction that recorded the consumption. That transaction is long, and recomputing inline held the shared rate row's X-lock for its whole length; doing it here holds that lock only for this consumer's own short transaction.

The command carries only the item's identity and the rate is recomputed from current state, so repeated commands coalesce and a redelivery recomputes the same absolute value.

func NewRecalcItemBurnRateConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	repos domain.RepoFactory,
	txManager db.TransactionManager[*sqlc.Queries, domain.RepoFactory],
) *RecalcItemBurnRateConsumer

func (c *RecalcItemBurnRateConsumer) Listen(ctx context.Context) error

type SalesOrderCreatedConsumer ¶

type SalesOrderCreatedConsumer struct {
	// contains filtered or unexported fields
}

SalesOrderCreatedConsumer processes sales-order-created events and runs the out-of-band side effects that should not block the create response — dispatching CRM sync for accounts with a connected integration (e.g. HubSpot), and warming the carrier transit estimate for the order's lane.

func NewSalesOrderCreatedConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	hubspotSync hubspotsync.Service,
	transitWarmer domain.TransitWarmer,
) *SalesOrderCreatedConsumer

func (c *SalesOrderCreatedConsumer) Listen(ctx context.Context) error

type SalesOrderShippingUpdatedConsumer ¶

type SalesOrderShippingUpdatedConsumer struct {
	// contains filtered or unexported fields
}

SalesOrderShippingUpdatedConsumer re-syncs an order's existing shipment records (carrier / service level / ship-to) after the order's shipping fields changed, out-of-band from the update response. This mirrors legacy updateCarrierByOrder / updateShipToByOrder, which cascaded these edits to shipments synchronously.

func NewSalesOrderShippingUpdatedConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	repos domain.RepoFactory,
	transitWarmer domain.TransitWarmer,
) *SalesOrderShippingUpdatedConsumer

func (c *SalesOrderShippingUpdatedConsumer) Listen(ctx context.Context) error

type SeedConsumer ¶

type SeedConsumer struct {
	// contains filtered or unexported fields
}

func NewSeedConsumer(rabbitmq messaging.MessageBroker, inboxRepo messaging.InboxRepo, seeder *repository.SandboxSeeder) *SeedConsumer

func (c *SeedConsumer) Listen(ctx context.Context) error

type SeedSandboxPayload ¶

type SeedSandboxPayload struct {
	AccountID string `json:"account_id"`
}

SeedSandboxPayload is the payload for CoreCmdSeedSandbox messages.

type SyncStripeCustomerConsumer ¶

type SyncStripeCustomerConsumer struct {
	// contains filtered or unexported fields
}

SyncStripeCustomerConsumer mirrors a customer create/update onto the account's connected Stripe integration.

The mutation itself already committed — this is Stripe catching up behind it. Running out-of-band is what keeps a Stripe outage from failing a customer edit that has nothing to do with payments.

func NewSyncStripeCustomerConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	stripeSync stripesync.Service,
) *SyncStripeCustomerConsumer

func (c *SyncStripeCustomerConsumer) Listen(ctx context.Context) error

type UndoBatchScanConsumer ¶

type UndoBatchScanConsumer struct {
	// contains filtered or unexported fields
}

UndoBatchScanConsumer reverses the inventory a scan recorded against a batch that has been deleted: the receipts it produced, the issues it consumed, and the reservations it drew down.

The delete itself already happened synchronously — this is the ledger catching up behind it, the mirror of ExecuteProductionStepConsumer running behind a scan.

func NewUndoBatchScanConsumer(
	rabbitmq messaging.MessageBroker,
	inboxRepo messaging.InboxRepo,
	repos domain.RepoFactory,
) *UndoBatchScanConsumer

func (c *UndoBatchScanConsumer) Listen(ctx context.Context) error