kra-oa/internal/integration

Integrations

internal/integration contains adapters that talk to systems outside the application process. They may open sockets, create SDK clients, keep reloadable state, or translate provider-specific protocols into biz interfaces.

Packages

• cache: Redis-backed cache with an in-memory fallback.
• email: SMTP email repository.
• mq: reloadable EMQX/MQTT client exposed through the shared pkg/mq interface.
• payment: payment-channel SDKs and callback/signature handling.
• storage: local and object-storage implementations of biz.FileStorage.
• websocket: reloadable Melody endpoint exposed through the shared pkg/websocket wrapper.

These packages are not utils: they perform I/O and own external dependency lifecycles. Their constructors are exposed through ProviderSet; cmd binds the cache.RedisProvider implementation to the shared data.Data container.

Stateless protocol helpers that do not own clients live in pkg/paymentkit. They are intentionally small and dependency light, while provider adapters remain here.

Shared WebSocket and MQ APIs

Business modules depend on websocket.Hub and mq.Client from the shared pkg/websocket and pkg/mq packages; they do not construct Melody or Paho clients and do not read system configuration directly. The system integration packages own runtime refresh and shutdown. Registered WebSocket handlers and MQTT subscriptions are retained when database-backed configuration replaces a live client.

ws.OnMessage(func(session *websocket.Session, payload []byte) {
	_ = ws.Send(session, payload)
})
_ = ws.Broadcast([]byte(`{"type":"system.ready"}`))

_ = mq.PublishJSON(ctx, mqClient, "orders/paid", event, mq.AtLeastOnce, false)
_ = mqClient.Subscribe(ctx, "orders/+/paid", mq.AtLeastOnce, handler)

Integration configuration is stored in sys_integration_configs:

• WebSocket: kind=websocket, provider=melody
• EMQX: kind=mq, provider=emqx

The YAML values are only migration/bootstrap inputs. After the integration table exists, the database is authoritative and runtime updates are applied without restarting the process. The WebSocket public path defaults to /ws.