wirabasalamah/Qris-Soundbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement phase 1 completion and phase 2 dynamic QR

Browse Source
This commit is contained in:
wirabasalamah
2026-05-26 08:06:48 +07:00
parent a152c99cce
commit 5624b92872
36 changed files with 3104 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
src/app.ts
View File

@ -15,7 +15,26 @@ import fs from "node:fs";
const app = express();
startNotificationOrchestrator();
app.use(helmet());
app.use(
helmet({
crossOriginResourcePolicy: {
policy: "cross-origin"
},
contentSecurityPolicy: {
directives: {
defaultSrc: ["'self'"],
scriptSrc: ["'self'", "'unsafe-inline'", "'unsafe-eval'", "https://cdn.tailwindcss.com"],
scriptSrcAttr: ["'unsafe-inline'"],
styleSrc: ["'self'", "'unsafe-inline'", "https://fonts.googleapis.com"],
fontSrc: ["'self'", "https://fonts.gstatic.com", "data:"],
imgSrc: ["'self'", "data:", "https://lh3.googleusercontent.com", "https://*.googleusercontent.com"],
connectSrc: ["'self'"],
objectSrc: ["'none'"],
baseUri: ["'self'"]
}
}
})
);
app.use(express.json());
app.use(morgan("dev"));
app.use(requestContext);

315
src/routes/admin.ts
View File

@ -56,6 +56,18 @@ import {
toNotificationPayload
} from "../shared/store/notificationStore";
import { retryNotificationByTransactionId } from "../shared/orchestrators/notificationOrchestrator";
import { createAuditLog, listAuditLogs, toAuditLogPayload } from "../shared/store/auditLogStore";
import { listLedgerEntries, toLedgerEntryPayload } from "../shared/store/ledgerStore";
import { resolveDeviceCapabilitySummary } from "../shared/services/deviceCapabilityResolver";
import {
getOrCreateDeviceConfig,
listDeviceConfigAcks,
toDeviceConfigAckPayload,
toDeviceConfigPayload,
upsertDeviceConfig
} from "../shared/store/deviceConfigStore";
import { listMqttMessages, toMqttMessagePayload, createMqttMessage } from "../shared/store/mqttMessageStore";
import { publishConfigPush } from "../shared/services/mqttPublisher";
const router = Router();
@ -104,6 +116,11 @@ type DeviceCommandInput = {
payload?: Record<string, unknown>;
};
type DeviceConfigInput = {
settings?: Record<string, unknown>;
config_version?: number;
};
type BindingInput = {
merchant_id?: string;
outlet_id?: string;
@ -262,6 +279,7 @@ async function buildDeviceAdminPayload(device: DeviceEntity) {
const latestHeartbeat = await getLatestHeartbeatByDeviceId(device.id);
return {
...toDevicePayload(device),
capability_summary: resolveDeviceCapabilitySummary(device),
derived_status: deriveDeviceStatus({
last_seen_at: device.last_seen_at,
network_strength: latestHeartbeat?.network_strength ?? null,
@ -327,6 +345,30 @@ function normalizeMerchantMode(payloadMode: MerchantCreateInput["payout_mode"]):
return payloadMode || "merchant_direct";
}
async function auditAdminAction(
req: Request,
payload: {
action: string;
entity_type: string;
entity_id: string;
before_json?: unknown;
after_json?: unknown;
}
) {
await createAuditLog({
actor_type: "admin",
actor_id: "admin",
action: payload.action,
entity_type: payload.entity_type,
entity_id: payload.entity_id,
before_json: payload.before_json,
after_json: payload.after_json,
source_ip: req.ip,
request_id: req.requestId,
trace_id: req.traceId
});
}
function validatePayoutConfig(payload: MerchantCreateInput) {
const mode = normalizeMerchantMode(payload.payout_mode);
if (mode === "merchant_direct") {
@ -430,6 +472,13 @@ router.post(
onboarding_status: payload.onboarding_status
});
await auditAdminAction(req, {
action: "merchant.create",
entity_type: "merchant",
entity_id: created.id,
after_json: toMerchantPayload(created)
});
res.status(201).json(successResponse(req, toMerchantPayload(created)));
}
);
@ -481,6 +530,13 @@ router.patch("/merchants/:merchantId", requireAdminToken, async (req: Request, r
}
const updated = await patchMerchant(req.params.merchantId, normalized);
await auditAdminAction(req, {
action: "merchant.update",
entity_type: "merchant",
entity_id: updated.id,
before_json: toMerchantPayload(existing),
after_json: toMerchantPayload(updated)
});
res.json(successResponse(req, toMerchantPayload(updated)));
});
@ -498,6 +554,14 @@ router.post("/merchants/:merchantId/approve", requireAdminToken, async (req: Req
onboarding_status: "approved"
});
await auditAdminAction(req, {
action: "merchant.approve",
entity_type: "merchant",
entity_id: updated.id,
before_json: toMerchantPayload(existing),
after_json: toMerchantPayload(updated)
});
res.json(successResponse(req, toMerchantPayload(updated)));
});
@ -520,6 +584,17 @@ router.post(
status: "inactive"
});
await auditAdminAction(req, {
action: "merchant.reject",
entity_type: "merchant",
entity_id: updated.id,
before_json: toMerchantPayload(existing),
after_json: {
...toMerchantPayload(updated),
rejection_reason: payload.reason
}
});
res.json(
successResponse(req, {
...toMerchantPayload(updated),
@ -591,7 +666,11 @@ router.post(
device_code: "DEV_SEED_A",
vendor: "seed-maker",
model: "v1",
communication_mode: "mqtt",
communication_mode: "static",
capability_profile_json: {
dynamic_qr: false,
flows: ["static_payment_notification"]
},
status: "active"
});
const deviceB = await createDevice({
@ -599,14 +678,28 @@ router.post(
vendor: "seed-maker",
model: "v1",
communication_mode: "mqtt",
capability_profile_json: {
dynamic_qr: {
mqtt: true,
api_direct: false
},
flows: ["dynamic_qr:mqtt", "static_payment_notification"]
},
status: "active"
});
const deviceC = await createDevice({
device_code: "DEV_SEED_C",
vendor: "seed-maker",
model: "v1",
communication_mode: "mqtt",
status: "inactive"
communication_mode: "api",
capability_profile_json: {
dynamic_qr: {
api_direct: true,
mqtt: false
},
flows: ["dynamic_qr:api_direct", "static_payment_notification"]
},
status: "active"
});
await bindDevice({
@ -732,6 +825,13 @@ router.post(
status: payload.status
});
await auditAdminAction(req, {
action: "outlet.create",
entity_type: "outlet",
entity_id: outlet.id,
after_json: toOutletPayload(outlet)
});
res.status(201).json(successResponse(req, outlet));
}
);
@ -775,7 +875,15 @@ router.patch("/outlets/:outletId", requireAdminToken, async (req: Request, res:
}
try {
const existing = await getOutletById(req.params.outletId);
const updated = await patchOutlet(req.params.outletId, payload);
await auditAdminAction(req, {
action: "outlet.update",
entity_type: "outlet",
entity_id: updated.id,
before_json: existing ? toOutletPayload(existing) : null,
after_json: toOutletPayload(updated)
});
res.json(successResponse(req, toOutletPayload(updated)));
} catch (err) {
if (err instanceof Error && err.message === "OUTLET_NOT_FOUND") {
@ -819,6 +927,13 @@ router.post(
status: payload.status
});
await auditAdminAction(req, {
action: "terminal.create",
entity_type: "terminal",
entity_id: terminal.id,
after_json: toTerminalPayload(terminal)
});
res.status(201).json(successResponse(req, toTerminalPayload(terminal)));
}
);
@ -865,7 +980,15 @@ router.patch("/terminals/:terminalId", requireAdminToken, async (req: Request, r
}
try {
const existing = await getTerminalById(req.params.terminalId);
const updated = await patchTerminal(req.params.terminalId, payload);
await auditAdminAction(req, {
action: "terminal.update",
entity_type: "terminal",
entity_id: updated.id,
before_json: existing ? toTerminalPayload(existing) : null,
after_json: toTerminalPayload(updated)
});
res.json(successResponse(req, toTerminalPayload(updated)));
} catch (err) {
if (err instanceof Error && err.message === "TERMINAL_NOT_FOUND") {
@ -894,6 +1017,12 @@ router.post("/devices", requireAdminToken, idempotency({ scope: "device.create",
}
const created = await createDevice(payload);
await auditAdminAction(req, {
action: "device.create",
entity_type: "device",
entity_id: created.id,
after_json: toDevicePayload(created)
});
res.status(201).json(successResponse(req, toDevicePayload(created)));
});
@ -969,7 +1098,7 @@ router.get("/devices/:deviceId", requireAdminToken, async (req: Request, res: Re
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
const activeBinding = await getActiveBindingByDevice(device.id);
const activeBinding = await getActiveBindingByDevice(device.id);
const latestHeartbeat = await getLatestHeartbeatByDeviceId(device.id);
const heartbeatCount24h = await getHeartbeatCountForDeviceLastHours(device.id);
const notifications = (await listNotificationsByDevice(device.id))
@ -980,6 +1109,7 @@ router.get("/devices/:deviceId", requireAdminToken, async (req: Request, res: Re
res.json(
successResponse(req, {
...toDevicePayload(device),
capability_summary: resolveDeviceCapabilitySummary(device),
derived_status: deriveDeviceStatus({
last_seen_at: device.last_seen_at,
network_strength: latestHeartbeat?.network_strength ?? null,
@ -1047,7 +1177,15 @@ router.patch("/devices/:deviceId", requireAdminToken, async (req: Request, res:
}
try {
const existing = await getDeviceById(req.params.deviceId);
const updated = await patchDevice(req.params.deviceId, payload);
await auditAdminAction(req, {
action: "device.update",
entity_type: "device",
entity_id: updated.id,
before_json: existing ? toDevicePayload(existing) : null,
after_json: toDevicePayload(updated)
});
res.json(successResponse(req, toDevicePayload(updated)));
} catch (err) {
if (err instanceof Error && err.message === "DEVICE_NOT_FOUND") {
@ -1094,6 +1232,13 @@ router.post("/devices/:deviceId/bind", requireAdminToken, idempotency({ scope: "
terminal_id: terminal.id
});
await auditAdminAction(req, {
action: "device.bind",
entity_type: "device_binding",
entity_id: binding.id,
after_json: toBindingPayload(binding)
});
res.json(successResponse(req, toBindingPayload(binding)));
});
@ -1108,6 +1253,14 @@ router.post("/devices/:deviceId/unbind", requireAdminToken, async (req: Request,
return next(new ApiError("BAD_REQUEST", "device has no active binding", 400));
}
await auditAdminAction(req, {
action: "device.unbind",
entity_type: "device_binding",
entity_id: binding.id,
before_json: toBindingPayload(binding),
after_json: toBindingPayload(binding)
});
res.json(successResponse(req, toBindingPayload(binding)));
});
@ -1199,6 +1352,99 @@ router.get("/devices/:deviceId/notifications", requireAdminToken, async (req: Re
res.json(successResponse(req, { device_id: device.id, notifications }));
});
router.get("/devices/:deviceId/config", requireAdminToken, async (req: Request, res: Response, next: NextFunction) => {
const device = await getDeviceById(req.params.deviceId);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
const config = await getOrCreateDeviceConfig(device.id);
const acks = (await listDeviceConfigAcks(device.id, 10)).map(toDeviceConfigAckPayload);
res.json(successResponse(req, { ...toDeviceConfigPayload(config), latest_acks: acks }));
});
router.patch("/devices/:deviceId/config", requireAdminToken, async (req: Request, res: Response, next: NextFunction) => {
const device = await getDeviceById(req.params.deviceId);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
const payload = req.body as DeviceConfigInput;
if (!payload || !payload.settings || typeof payload.settings !== "object") {
return next(new ApiError("BAD_REQUEST", "settings object is required", 400));
}
const config = await upsertDeviceConfig({
device_id: device.id,
settings_json: payload.settings,
config_version: payload.config_version
});
const mqttPayload = {
message_type: "config_push" as const,
config_version: config.config_version,
settings: config.settings_json
};
const publishResult = await publishConfigPush(device.id, mqttPayload);
const outbox = await createMqttMessage({
direction: "downlink",
device_id: device.id,
topic: publishResult.topic,
message_type: "config_push",
correlation_id: `config:${config.config_version}`,
payload_json: mqttPayload,
publish_status: publishResult.ok ? "sent" : "failed",
reason: publishResult.reason
});
await auditAdminAction(req, {
action: "device.config_push",
entity_type: "device",
entity_id: device.id,
after_json: {
config,
downlink_message_id: outbox.id
}
});
res.json(
successResponse(req, {
config: toDeviceConfigPayload(config),
downlink_message: toMqttMessagePayload(outbox)
})
);
});
router.get("/devices/:deviceId/mqtt-messages", requireAdminToken, async (req: Request, res: Response, next: NextFunction) => {
const device = await getDeviceById(req.params.deviceId);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
const directionRaw = (req.query.direction as string | undefined)?.trim();
const direction = directionRaw === "uplink" || directionRaw === "downlink" ? directionRaw : undefined;
if (directionRaw && !direction) {
return next(new ApiError("BAD_REQUEST", "direction must be uplink or downlink", 400));
}
const messageType = (req.query.message_type as string | undefined)?.trim();
const correlationId = (req.query.correlation_id as string | undefined)?.trim();
const limitRaw = req.query.limit as string | undefined;
const limit = limitRaw ? Number(limitRaw) : 100;
if (!Number.isFinite(limit) || limit <= 0) {
return next(new ApiError("BAD_REQUEST", "limit must be a positive number", 400));
}
const messages = await listMqttMessages({
device_id: device.id,
direction,
message_type: messageType || undefined,
correlation_id: correlationId || undefined,
limit
});
res.json(successResponse(req, { device_id: device.id, messages: messages.map(toMqttMessagePayload) }));
});
router.post(
"/transactions",
requireAdminToken,
@ -1268,6 +1514,13 @@ router.post(
status: payload.status || "initiated"
});
await auditAdminAction(req, {
action: "transaction.create",
entity_type: "transaction",
entity_id: created.id,
after_json: toTransactionPayload(created)
});
res.status(201).json(successResponse(req, toTransactionPayload(created)));
}
);
@ -1331,6 +1584,7 @@ router.get(
}
const events = (await getTransactionEvents(tx.id)).map(toTransactionEventPayload);
const ledger_entries = (await listLedgerEntries({ transaction_id: tx.id })).map(toLedgerEntryPayload);
const bindingByTerminal = tx.terminal_id ? await getActiveBindingByTerminal(tx.terminal_id) : null;
const heartbeatDeviceId = tx.device_id || bindingByTerminal?.device_id;
const heartbeatHistory = heartbeatDeviceId
@ -1350,6 +1604,7 @@ router.get(
successResponse(req, {
transaction: toTransactionPayload(tx),
events,
ledger_entries,
heartbeat_device_id: heartbeatDeviceId,
heartbeat_history: heartbeatHistory
})
@ -1588,4 +1843,56 @@ router.get("/notifications/failed", requireAdminToken, async (req: Request, res:
res.json(successResponse(req, filtered));
});
router.get("/audit-logs", requireAdminToken, async (req: Request, res: Response, next: NextFunction) => {
const entityType = (req.query.entity_type as string | undefined)?.trim();
const entityId = (req.query.entity_id as string | undefined)?.trim();
const action = (req.query.action as string | undefined)?.trim();
const from = req.query.from as string | undefined;
const to = req.query.to as string | undefined;
const limitRaw = req.query.limit as string | undefined;
const limit = limitRaw ? Number(limitRaw) : 100;
if (from && Number.isNaN(Date.parse(from))) {
return next(new ApiError("BAD_REQUEST", "from must be valid ISO datetime", 400));
}
if (to && Number.isNaN(Date.parse(to))) {
return next(new ApiError("BAD_REQUEST", "to must be valid ISO datetime", 400));
}
if (!Number.isFinite(limit) || limit <= 0) {
return next(new ApiError("BAD_REQUEST", "limit must be a positive number", 400));
}
const logs = await listAuditLogs({
entity_type: entityType || undefined,
entity_id: entityId || undefined,
action: action || undefined,
from,
to,
limit
});
res.json(successResponse(req, logs.map(toAuditLogPayload)));
});
router.get("/ledger-entries", requireAdminToken, async (req: Request, res: Response, next: NextFunction) => {
const transactionId = (req.query.transaction_id as string | undefined)?.trim();
const merchantId = (req.query.merchant_id as string | undefined)?.trim();
const limitRaw = req.query.limit as string | undefined;
const limit = limitRaw ? Number(limitRaw) : 100;
if (!Number.isFinite(limit) || limit <= 0) {
return next(new ApiError("BAD_REQUEST", "limit must be a positive number", 400));
}
const entries = await listLedgerEntries({
transaction_id: transactionId || undefined,
merchant_id: merchantId || undefined,
limit
});
res.json(successResponse(req, entries.map(toLedgerEntryPayload)));
});
export default router;

294
src/routes/device.ts
View File

@ -5,6 +5,20 @@ import { successResponse } from "../shared/middleware/errorMiddleware";
import { getDeviceById, patchDevice } from "../shared/store/deviceStore";
import { createDeviceHeartbeat } from "../shared/store/heartbeatStore";
import { acknowledgeDeviceCommand } from "../shared/store/deviceCommandStore";
import { getActiveBindingByDevice } from "../shared/store/bindingStore";
import { getTerminalById } from "../shared/store/locationStore";
import { readIdempotency, writeIdempotency } from "../shared/idempotency/idempotencyStore";
import { env } from "../config/env";
import { supportsDynamicQrFlow } from "../shared/services/deviceCapabilityResolver";
import { createDynamicQrTransaction } from "../shared/services/dynamicQrOrchestrator";
import { createMqttMessage } from "../shared/store/mqttMessageStore";
import { publishDynamicQrResponse } from "../shared/services/mqttPublisher";
import {
createDeviceConfigAck,
getOrCreateDeviceConfig,
toDeviceConfigAckPayload,
toDeviceConfigPayload
} from "../shared/store/deviceConfigStore";
const router = Router();
@ -25,6 +39,28 @@ type CommandAckInput = {
result_payload?: Record<string, unknown>;
};
type DynamicQrInput = {
device_id?: string;
terminal_id?: string;
amount?: unknown;
currency?: string;
request_id?: string;
expires_in_seconds?: unknown;
};
type MqttDynamicQrInput = DynamicQrInput & {
message_type?: string;
created_at?: string;
};
type ConfigAckInput = {
device_id?: string;
config_version?: unknown;
status?: "applied" | "failed";
reason?: string;
result_payload?: Record<string, unknown>;
};
function normalizeNumberOrNull(value: unknown): number | null {
if (typeof value === "string") {
const parsed = Number(value);
@ -40,6 +76,35 @@ function normalizeNumberOrNull(value: unknown): number | null {
return null;
}
function normalizePositiveAmount(value: unknown): number | null {
const normalized = normalizeNumberOrNull(value);
if (normalized === null || normalized <= 0) {
return null;
}
return normalized;
}
function normalizeTtl(value: unknown): number | undefined {
if (value === undefined || value === null || value === "") {
return undefined;
}
const normalized = normalizeNumberOrNull(value);
if (normalized === null || normalized <= 0) {
return undefined;
}
return normalized;
}
function normalizePositiveInteger(value: unknown): number | null {
const normalized = normalizeNumberOrNull(value);
if (normalized === null || normalized <= 0 || !Number.isInteger(normalized)) {
return null;
}
return normalized;
}
function normalizeSignalStrength(value: unknown): number | null {
const normalized = normalizeNumberOrNull(value);
if (normalized === null) {
@ -165,4 +230,233 @@ router.post("/commands/ack", requireDeviceToken, async (req: Request, res: Respo
);
});
router.post("/transactions/dynamic-qr", requireDeviceToken, async (req: Request, res: Response, next: NextFunction) => {
const payload = req.body as DynamicQrInput;
if (!payload || !payload.device_id || !payload.terminal_id || !payload.request_id) {
return next(new ApiError("BAD_REQUEST", "device_id, terminal_id, request_id are required", 400));
}
const amount = normalizePositiveAmount(payload.amount);
if (amount === null) {
return next(new ApiError("INVALID_AMOUNT", "amount must be a positive number", 400));
}
const currency = payload.currency && payload.currency.trim() ? payload.currency.trim().toUpperCase() : "IDR";
if (currency !== "IDR") {
return next(new ApiError("BAD_REQUEST", "currency must be IDR for QRIS dynamic MVP", 400));
}
const idempotencyKey = req.header("Idempotency-Key") || payload.request_id;
const cached = readIdempotency("device.dynamic_qr.create", idempotencyKey);
if (cached) {
return res.json(successResponse(req, (cached as { data: unknown }).data ?? cached));
}
const device = await getDeviceById(payload.device_id);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
if (device.status !== "active") {
return next(new ApiError("DEVICE_CAPABILITY_NOT_SUPPORTED", "inactive device cannot create dynamic QR", 400));
}
if (!supportsDynamicQrFlow(device, "api_direct")) {
return next(new ApiError("DEVICE_CAPABILITY_NOT_SUPPORTED", "device does not support API-direct dynamic QR", 400));
}
const terminal = await getTerminalById(payload.terminal_id);
if (!terminal) {
return next(new ApiError("NOT_FOUND", "terminal not found", 404));
}
if (terminal.qr_mode !== "dynamic_api") {
return next(new ApiError("DEVICE_CAPABILITY_NOT_SUPPORTED", "terminal is not configured for API dynamic QR", 400));
}
const binding = await getActiveBindingByDevice(device.id);
if (!binding || binding.terminal_id !== terminal.id) {
return next(new ApiError("DEVICE_NOT_BOUND", "device is not actively bound to requested terminal", 400));
}
const created = await createDynamicQrTransaction({
request_id: payload.request_id,
device_id: device.id,
merchant_id: binding.merchant_id,
outlet_id: binding.outlet_id,
terminal_id: binding.terminal_id,
amount,
currency,
expires_in_seconds: normalizeTtl(payload.expires_in_seconds)
});
const responseData = {
...created,
request_id: payload.request_id
};
writeIdempotency(
"device.dynamic_qr.create",
idempotencyKey,
{ data: responseData },
env.IDEMPOTENCY_TTL_MS
);
res.status(201).json(successResponse(req, responseData));
});
router.post("/mqtt/uplink/dynamic-qr/request", requireDeviceToken, async (req: Request, res: Response, next: NextFunction) => {
const payload = req.body as MqttDynamicQrInput;
if (!payload || !payload.device_id || !payload.terminal_id || !payload.request_id) {
return next(new ApiError("BAD_REQUEST", "device_id, terminal_id, request_id are required", 400));
}
if (payload.message_type && payload.message_type !== "dynamic_qr_request") {
return next(new ApiError("BAD_REQUEST", "message_type must be dynamic_qr_request", 400));
}
const amount = normalizePositiveAmount(payload.amount);
if (amount === null) {
return next(new ApiError("INVALID_AMOUNT", "amount must be a positive number", 400));
}
const currency = payload.currency && payload.currency.trim() ? payload.currency.trim().toUpperCase() : "IDR";
if (currency !== "IDR") {
return next(new ApiError("BAD_REQUEST", "currency must be IDR for QRIS dynamic MVP", 400));
}
const cached = readIdempotency("device.dynamic_qr.mqtt", payload.request_id);
if (cached) {
return res.json(successResponse(req, (cached as { data: unknown }).data ?? cached));
}
const device = await getDeviceById(payload.device_id);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
if (device.status !== "active" || !supportsDynamicQrFlow(device, "mqtt")) {
return next(new ApiError("DEVICE_CAPABILITY_NOT_SUPPORTED", "device does not support MQTT dynamic QR", 400));
}
const terminal = await getTerminalById(payload.terminal_id);
if (!terminal) {
return next(new ApiError("NOT_FOUND", "terminal not found", 404));
}
if (terminal.qr_mode !== "dynamic_mqtt") {
return next(new ApiError("DEVICE_CAPABILITY_NOT_SUPPORTED", "terminal is not configured for MQTT dynamic QR", 400));
}
const binding = await getActiveBindingByDevice(device.id);
if (!binding || binding.terminal_id !== terminal.id) {
return next(new ApiError("DEVICE_NOT_BOUND", "device is not actively bound to requested terminal", 400));
}
await createMqttMessage({
direction: "uplink",
device_id: device.id,
topic: `devices/${device.id}/uplink/dynamic-qr/request`,
message_type: "dynamic_qr_request",
correlation_id: payload.request_id,
payload_json: {
...payload,
amount,
currency
}
});
const created = await createDynamicQrTransaction({
request_id: payload.request_id,
device_id: device.id,
merchant_id: binding.merchant_id,
outlet_id: binding.outlet_id,
terminal_id: binding.terminal_id,
amount,
currency,
expires_in_seconds: normalizeTtl(payload.expires_in_seconds),
initiation_mode: "dynamic_mqtt"
});
const mqttPayload = {
message_type: "dynamic_qr_response" as const,
correlation_id: payload.request_id,
transaction_id: created.transaction_id,
status: "success" as const,
qr_payload: created.qr_payload,
expires_at: created.expires_at
};
const publishResult = await publishDynamicQrResponse(device.id, mqttPayload);
const outbox = await createMqttMessage({
direction: "downlink",
device_id: device.id,
topic: publishResult.topic,
message_type: "dynamic_qr_response",
correlation_id: payload.request_id,
payload_json: mqttPayload,
publish_status: publishResult.ok ? "sent" : "failed",
reason: publishResult.reason
});
const responseData = {
correlation_id: payload.request_id,
transaction_id: created.transaction_id,
status: "success",
qr_payload: created.qr_payload,
expires_at: created.expires_at,
downlink_message_id: outbox.id,
publish_status: outbox.publish_status,
partner_reference: created.partner_reference
};
writeIdempotency("device.dynamic_qr.mqtt", payload.request_id, { data: responseData }, env.IDEMPOTENCY_TTL_MS);
res.status(201).json(successResponse(req, responseData));
});
router.get("/config", requireDeviceToken, async (req: Request, res: Response, next: NextFunction) => {
const deviceId = (req.query.device_id as string | undefined) || (req.body as { device_id?: string } | undefined)?.device_id;
if (!deviceId) {
return next(new ApiError("BAD_REQUEST", "device_id is required", 400));
}
const device = await getDeviceById(deviceId);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
const config = await getOrCreateDeviceConfig(device.id);
res.json(successResponse(req, toDeviceConfigPayload(config)));
});
router.post("/config/ack", requireDeviceToken, async (req: Request, res: Response, next: NextFunction) => {
const payload = req.body as ConfigAckInput;
if (!payload || !payload.device_id || !payload.status) {
return next(new ApiError("BAD_REQUEST", "device_id, status are required", 400));
}
if (!["applied", "failed"].includes(payload.status)) {
return next(new ApiError("BAD_REQUEST", "status must be applied or failed", 400));
}
const configVersion = normalizePositiveInteger(payload.config_version);
if (configVersion === null) {
return next(new ApiError("BAD_REQUEST", "config_version must be a positive integer", 400));
}
const device = await getDeviceById(payload.device_id);
if (!device) {
return next(new ApiError("NOT_FOUND", "device not found", 404));
}
const ack = await createDeviceConfigAck({
device_id: device.id,
config_version: configVersion,
status: payload.status,
reason: payload.reason,
payload_json: payload.result_payload || {}
});
res.json(successResponse(req, toDeviceConfigAckPayload(ack)));
});
export default router;

44
src/routes/integrations.ts
View File

@ -11,6 +11,7 @@ import {
} from "../shared/store/transactionStore";
import { emitTransactionPaid } from "../shared/events/transactionEvents";
import { env } from "../config/env";
import { createAuditLog } from "../shared/store/auditLogStore";
const router = Router();
@ -159,6 +160,29 @@ function writeCallbackResult(idempotencyKey: string, response: CallbackResponse,
);
}
async function auditWebhookAction(
req: Request,
payload: {
action: string;
entity_id: string;
before_json?: unknown;
after_json?: unknown;
}
) {
await createAuditLog({
actor_type: "webhook",
actor_id: "qris_partner",
action: payload.action,
entity_type: "transaction",
entity_id: payload.entity_id,
before_json: payload.before_json,
after_json: payload.after_json,
source_ip: req.ip,
request_id: req.requestId,
trace_id: req.traceId
});
}
async function makeResponseEventId(txId: string, fallbackTag: string) {
const events = await getTransactionEvents(txId);
return events.at(-1)?.id || `${fallbackTag}_${Date.now()}`;
@ -281,6 +305,12 @@ router.post("/qris/callback", async (req: Request, res: Response, next: NextFunc
}
if (!wasPaid) {
await auditWebhookAction(req, {
action: "transaction.mark_paid",
entity_id: updated.id,
before_json: tx,
after_json: updated
});
emitTransactionPaid({
transaction_id: updated.id,
merchant_id: updated.merchant_id,
@ -333,6 +363,13 @@ router.post("/qris/callback", async (req: Request, res: Response, next: NextFunc
throw error;
}
await auditWebhookAction(req, {
action: "transaction.mark_expired",
entity_id: updated.id,
before_json: tx,
after_json: updated
});
const response = buildCallbackResponse(req, updated.id, await makeResponseEventId(updated.id, "tx_event"));
writeCallbackResult(idempotencyKey, response, updated.id);
return res.json(successResponse(req, response));
@ -359,6 +396,13 @@ router.post("/qris/callback", async (req: Request, res: Response, next: NextFunc
throw error;
}
await auditWebhookAction(req, {
action: "transaction.mark_failed",
entity_id: updated.id,
before_json: tx,
after_json: updated
});
const response = buildCallbackResponse(
req,
updated.id,

95
src/shared/db/pool.ts
View File

@ -154,6 +154,41 @@ CREATE TABLE IF NOT EXISTS device_commands (
CREATE INDEX IF NOT EXISTS idx_device_commands_device_request ON device_commands (device_id, requested_at DESC);
CREATE TABLE IF NOT EXISTS mqtt_messages (
id TEXT PRIMARY KEY,
direction TEXT NOT NULL CHECK (direction IN ('uplink', 'downlink')),
device_id TEXT NOT NULL REFERENCES devices (id) ON DELETE CASCADE,
topic TEXT NOT NULL,
message_type TEXT NOT NULL,
correlation_id TEXT,
payload_json JSONB NOT NULL DEFAULT '{}'::jsonb,
publish_status TEXT NOT NULL DEFAULT 'recorded' CHECK (publish_status IN ('recorded', 'sent', 'failed')),
reason TEXT,
created_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_mqtt_messages_device_created ON mqtt_messages (device_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_mqtt_messages_correlation ON mqtt_messages (correlation_id);
CREATE TABLE IF NOT EXISTS device_configs (
device_id TEXT PRIMARY KEY REFERENCES devices (id) ON DELETE CASCADE,
config_version INT NOT NULL,
settings_json JSONB NOT NULL DEFAULT '{}'::jsonb,
updated_at TIMESTAMPTZ NOT NULL
);
CREATE TABLE IF NOT EXISTS device_config_acks (
id TEXT PRIMARY KEY,
device_id TEXT NOT NULL REFERENCES devices (id) ON DELETE CASCADE,
config_version INT NOT NULL,
status TEXT NOT NULL CHECK (status IN ('applied', 'failed')),
reason TEXT,
payload_json JSONB NOT NULL DEFAULT '{}'::jsonb,
acked_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_device_config_acks_device ON device_config_acks (device_id, acked_at DESC);
CREATE TABLE IF NOT EXISTS transactions (
id TEXT PRIMARY KEY,
transaction_code TEXT NOT NULL UNIQUE,
@ -211,5 +246,65 @@ CREATE TABLE IF NOT EXISTS notifications (
CREATE INDEX IF NOT EXISTS idx_notifications_device_status ON notifications (device_id, delivery_status);
CREATE INDEX IF NOT EXISTS idx_notifications_status_created ON notifications (delivery_status, created_at DESC);
CREATE TABLE IF NOT EXISTS ledger_entries (
id TEXT PRIMARY KEY,
transaction_id TEXT NOT NULL REFERENCES transactions (id) ON DELETE CASCADE,
merchant_id TEXT NOT NULL REFERENCES merchants (id) ON DELETE CASCADE,
entry_type TEXT NOT NULL CHECK (entry_type IN ('gross_income', 'platform_fee', 'merchant_payable')),
amount NUMERIC(20,2) NOT NULL,
currency TEXT NOT NULL DEFAULT 'IDR',
direction TEXT NOT NULL CHECK (direction IN ('credit', 'debit')),
status TEXT NOT NULL DEFAULT 'posted' CHECK (status IN ('posted', 'voided')),
metadata_json JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL,
CONSTRAINT ledger_entries_unique_tx_type UNIQUE (transaction_id, entry_type)
);
CREATE INDEX IF NOT EXISTS idx_ledger_entries_merchant_created ON ledger_entries (merchant_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_ledger_entries_tx ON ledger_entries (transaction_id);
CREATE TABLE IF NOT EXISTS roles (
id TEXT PRIMARY KEY,
name TEXT NOT NULL UNIQUE,
permissions_json JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL
);
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
role_id TEXT NOT NULL REFERENCES roles (id),
status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'inactive')),
created_at TIMESTAMPTZ NOT NULL
);
CREATE TABLE IF NOT EXISTS audit_logs (
id TEXT PRIMARY KEY,
actor_type TEXT NOT NULL,
actor_id TEXT,
action TEXT NOT NULL,
entity_type TEXT NOT NULL,
entity_id TEXT NOT NULL,
before_json JSONB,
after_json JSONB,
source_ip TEXT,
request_id TEXT,
trace_id TEXT,
created_at TIMESTAMPTZ NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_audit_logs_entity ON audit_logs (entity_type, entity_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_audit_logs_action ON audit_logs (action, created_at DESC);
INSERT INTO roles (id, name, permissions_json, created_at)
VALUES ('role_admin', 'admin', '{"admin":"*"}'::jsonb, NOW())
ON CONFLICT (id) DO NOTHING;
INSERT INTO users (id, name, email, password_hash, role_id, status, created_at)
VALUES ('user_admin_seed', 'Admin Seed', 'admin@example.local', 'dev-only-admin-password', 'role_admin', 'active', NOW())
ON CONFLICT (id) DO NOTHING;
COMMIT;
`;

5
src/shared/errors/index.ts
View File

@ -17,7 +17,10 @@ export type ErrorCode =
| "NOTIFICATION_DEVICE_UNAVAILABLE"
| "NOTIFICATION_NO_ACTIVE_BINDING"
| "NOTIFICATION_PUBLISH_FAILED"
| "NOTIFICATION_RETRY_EXHAUSTED";
| "NOTIFICATION_RETRY_EXHAUSTED"
| "INVALID_AMOUNT"
| "DEVICE_NOT_BOUND"
| "DEVICE_CAPABILITY_NOT_SUPPORTED";
export interface ApiErrorShape {
code: ErrorCode;

45
src/shared/services/deviceCapabilityResolver.ts Normal file
View File

@ -0,0 +1,45 @@
import type { DeviceEntity } from "../store/deviceStore";
type CapabilityProfile = {
dynamic_qr?: boolean | {
api_direct?: boolean;
mqtt?: boolean;
};
flows?: string[];
};
export type DynamicQrFlow = "api_direct" | "mqtt";
function getProfile(device: DeviceEntity): CapabilityProfile {
return (device.capability_profile_json || {}) as CapabilityProfile;
}
export function supportsDynamicQrFlow(device: DeviceEntity, flow: DynamicQrFlow): boolean {
const profile = getProfile(device);
const flows = Array.isArray(profile.flows) ? profile.flows : [];
if (flow === "api_direct" && device.communication_mode !== "api") {
return false;
}
if (flow === "mqtt" && device.communication_mode !== "mqtt") {
return false;
}
if (typeof profile.dynamic_qr === "boolean") {
return profile.dynamic_qr || flows.includes(`dynamic_qr:${flow}`);
}
if (profile.dynamic_qr && typeof profile.dynamic_qr === "object") {
return Boolean(profile.dynamic_qr[flow]) || flows.includes(`dynamic_qr:${flow}`);
}
return flows.includes(`dynamic_qr:${flow}`);
}
export function resolveDeviceCapabilitySummary(device: DeviceEntity) {
return {
dynamic_qr_api_direct: supportsDynamicQrFlow(device, "api_direct"),
dynamic_qr_mqtt: supportsDynamicQrFlow(device, "mqtt")
};
}

104
src/shared/services/dynamicQrOrchestrator.ts Normal file
View File

@ -0,0 +1,104 @@
import { randomUUID } from "node:crypto";
import { createTransaction, addTransactionEvent, toTransactionPayload } from "../store/transactionStore";
export type DynamicQrCreateResult = {
request_id: string;
correlation_id: string;
transaction_id: string;
transaction_code: string;
qr_type: "dynamic";
qr_payload: string;
expires_at: string;
status: "awaiting_payment";
partner_reference: string;
};
function makePartnerReference(requestId: string) {
const clean = requestId.replace(/[^a-zA-Z0-9_-]/g, "").slice(0, 40);
return `DYN-${clean || randomUUID().slice(0, 12)}`;
}
function makeDynamicQrPayload(input: {
transactionId: string;
partnerReference: string;
amount: number;
currency: string;
expiresAt: string;
}) {
const amountMinor = Math.round(input.amount * 100);
const encoded = Buffer.from(
JSON.stringify({
type: "QRIS_DYNAMIC_MOCK",
transaction_id: input.transactionId,
partner_reference: input.partnerReference,
amount_minor: amountMinor,
currency: input.currency,
expires_at: input.expiresAt
})
).toString("base64url");
return `QRIS-DYNAMIC-MOCK.${encoded}`;
}
export async function createDynamicQrTransaction(input: {
request_id: string;
device_id: string;
merchant_id: string;
outlet_id: string;
terminal_id: string;
amount: number;
currency?: string;
expires_in_seconds?: number;
initiation_mode?: "dynamic_api" | "dynamic_mqtt";
}): Promise<DynamicQrCreateResult> {
const ttlSeconds = Math.min(Math.max(input.expires_in_seconds || 300, 60), 1800);
const expiresAt = new Date(Date.now() + ttlSeconds * 1000).toISOString();
const partnerReference = makePartnerReference(input.request_id);
const tx = await createTransaction({
merchant_id: input.merchant_id,
outlet_id: input.outlet_id,
terminal_id: input.terminal_id,
device_id: input.device_id,
partner_reference: partnerReference,
amount: input.amount,
currency: input.currency || "IDR",
qr_mode: "dynamic",
initiation_mode: input.initiation_mode || "dynamic_api",
status: "awaiting_payment",
expired_at: expiresAt
});
const qrPayload = makeDynamicQrPayload({
transactionId: tx.id,
partnerReference,
amount: tx.amount,
currency: tx.currency,
expiresAt
});
await addTransactionEvent({
transaction_id: tx.id,
event_type: "DYNAMIC_QR_CREATED",
source: "device",
payload_json: {
request_id: input.request_id,
correlation_id: input.request_id,
device_id: input.device_id,
qr_payload: qrPayload,
expires_at: expiresAt,
transaction: toTransactionPayload(tx)
}
});
return {
request_id: input.request_id,
correlation_id: input.request_id,
transaction_id: tx.id,
transaction_code: tx.transaction_code,
qr_type: "dynamic",
qr_payload: qrPayload,
expires_at: expiresAt,
status: "awaiting_payment",
partner_reference: partnerReference
};
}

50
src/shared/services/mqttPublisher.ts
View File

@ -15,14 +15,29 @@ type PaymentSuccessPayload = {
display_text: string;
};
export type MqttPublishResult = {
type DynamicQrResponsePayload = {
message_type: "dynamic_qr_response";
correlation_id: string;
transaction_id: string;
status: "success";
qr_payload: string;
expires_at: string;
};
type ConfigPushPayload = {
message_type: "config_push";
config_version: number;
settings: Record<string, unknown>;
};
export type MqttPublishResult<TPayload = PaymentSuccessPayload> = {
ok: boolean;
topic: string;
qos: 1;
retained: false;
publishedAt: string;
reason?: string;
payload: PaymentSuccessPayload;
payload: TPayload;
};
const forcedFailAll = String(env.MQTT_PUBLISH_FORCE_FAIL_ALL).toLowerCase() === "true";
@ -72,11 +87,22 @@ export function makePaymentSuccessTopic(deviceId: string) {
return `devices/${deviceId}/downlink/payment/success`;
}
export async function publishPaymentSuccess(payload: PaymentSuccessPayload): Promise<MqttPublishResult> {
const publishedAt = new Date().toISOString();
const topic = makePaymentSuccessTopic(payload.device_id);
export function makeDynamicQrResponseTopic(deviceId: string) {
return `devices/${deviceId}/downlink/dynamic-qr/response`;
}
if (shouldForceFail(payload.device_id)) {
export function makeConfigPushTopic(deviceId: string) {
return `devices/${deviceId}/downlink/config/push`;
}
async function publishMqttPayload<TPayload>(
deviceId: string,
topic: string,
payload: TPayload
): Promise<MqttPublishResult<TPayload>> {
const publishedAt = new Date().toISOString();
if (shouldForceFail(deviceId)) {
return {
ok: false,
topic,
@ -97,3 +123,15 @@ export async function publishPaymentSuccess(payload: PaymentSuccessPayload): Pro
payload
};
}
export async function publishPaymentSuccess(payload: PaymentSuccessPayload): Promise<MqttPublishResult> {
return publishMqttPayload(payload.device_id, makePaymentSuccessTopic(payload.device_id), payload);
}
export async function publishDynamicQrResponse(deviceId: string, payload: DynamicQrResponsePayload) {
return publishMqttPayload(deviceId, makeDynamicQrResponseTopic(deviceId), payload);
}
export async function publishConfigPush(deviceId: string, payload: ConfigPushPayload) {
return publishMqttPayload(deviceId, makeConfigPushTopic(deviceId), payload);
}

136
src/shared/store/auditLogStore.ts Normal file
View File

@ -0,0 +1,136 @@
import { randomUUID } from "node:crypto";
import { getPool } from "../db/pool";
export interface AuditLogEntity {
id: string;
actor_type: "admin" | "device" | "webhook" | "system";
actor_id?: string;
action: string;
entity_type: string;
entity_id: string;
before_json?: Record<string, unknown> | null;
after_json?: Record<string, unknown> | null;
source_ip?: string;
request_id?: string;
trace_id?: string;
created_at: string;
}
function nowIso() {
return new Date().toISOString();
}
function mapAuditLog(row: any): AuditLogEntity {
return {
id: row.id,
actor_type: row.actor_type,
actor_id: row.actor_id || undefined,
action: row.action,
entity_type: row.entity_type,
entity_id: row.entity_id,
before_json: row.before_json || null,
after_json: row.after_json || null,
source_ip: row.source_ip || undefined,
request_id: row.request_id || undefined,
trace_id: row.trace_id || undefined,
created_at: row.created_at
};
}
export async function createAuditLog(payload: {
actor_type: AuditLogEntity["actor_type"];
actor_id?: string;
action: string;
entity_type: string;
entity_id: string;
before_json?: unknown;
after_json?: unknown;
source_ip?: string;
request_id?: string;
trace_id?: string;
}): Promise<AuditLogEntity> {
const { rows } = await getPool().query(
`INSERT INTO audit_logs (
id,
actor_type,
actor_id,
action,
entity_type,
entity_id,
before_json,
after_json,
source_ip,
request_id,
trace_id,
created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
RETURNING *`,
[
randomUUID(),
payload.actor_type,
payload.actor_id || null,
payload.action,
payload.entity_type,
payload.entity_id,
payload.before_json || null,
payload.after_json || null,
payload.source_ip || null,
payload.request_id || null,
payload.trace_id || null,
nowIso()
]
);
return mapAuditLog(rows[0]);
}
export async function listAuditLogs(filter?: {
entity_type?: string;
entity_id?: string;
action?: string;
from?: string;
to?: string;
limit?: number;
}): Promise<AuditLogEntity[]> {
const clauses: string[] = [];
const params: unknown[] = [];
let i = 1;
if (filter?.entity_type) {
clauses.push(`entity_type = $${i++}`);
params.push(filter.entity_type);
}
if (filter?.entity_id) {
clauses.push(`entity_id = $${i++}`);
params.push(filter.entity_id);
}
if (filter?.action) {
clauses.push(`action = $${i++}`);
params.push(filter.action);
}
if (filter?.from) {
clauses.push(`created_at >= $${i++}`);
params.push(filter.from);
}
if (filter?.to) {
clauses.push(`created_at <= $${i++}`);
params.push(filter.to);
}
const limit = Math.min(Math.max(filter?.limit || 100, 1), 500);
const where = clauses.length ? `WHERE ${clauses.join(" AND ")}` : "";
const { rows } = await getPool().query(
`SELECT * FROM audit_logs ${where} ORDER BY created_at DESC LIMIT ${limit}`,
params
);
return rows.map(mapAuditLog);
}
export function toAuditLogPayload(auditLog: AuditLogEntity) {
return { ...auditLog };
}

140
src/shared/store/deviceConfigStore.ts Normal file
View File

@ -0,0 +1,140 @@
import { randomUUID } from "node:crypto";
import { getPool } from "../db/pool";
export interface DeviceConfigEntity {
device_id: string;
config_version: number;
settings_json: Record<string, unknown>;
updated_at: string;
}
export interface DeviceConfigAckEntity {
id: string;
device_id: string;
config_version: number;
status: "applied" | "failed";
reason?: string;
payload_json: Record<string, unknown>;
acked_at: string;
}
const DEFAULT_SETTINGS = {
volume: 80,
language: "id-ID",
heartbeat_interval_seconds: 60
};
function nowIso() {
return new Date().toISOString();
}
function mapConfig(row: any): DeviceConfigEntity {
return {
device_id: row.device_id,
config_version: Number(row.config_version),
settings_json: row.settings_json || {},
updated_at: row.updated_at
};
}
function mapAck(row: any): DeviceConfigAckEntity {
return {
id: row.id,
device_id: row.device_id,
config_version: Number(row.config_version),
status: row.status,
reason: row.reason || undefined,
payload_json: row.payload_json || {},
acked_at: row.acked_at
};
}
export async function getDeviceConfig(deviceId: string): Promise<DeviceConfigEntity | null> {
const { rows } = await getPool().query("SELECT * FROM device_configs WHERE device_id = $1", [deviceId]);
return rows[0] ? mapConfig(rows[0]) : null;
}
export async function getOrCreateDeviceConfig(deviceId: string): Promise<DeviceConfigEntity> {
const existing = await getDeviceConfig(deviceId);
if (existing) {
return existing;
}
return upsertDeviceConfig({
device_id: deviceId,
settings_json: DEFAULT_SETTINGS
});
}
export async function upsertDeviceConfig(payload: {
device_id: string;
settings_json: Record<string, unknown>;
config_version?: number;
}): Promise<DeviceConfigEntity> {
const existing = await getDeviceConfig(payload.device_id);
const nextVersion = payload.config_version || (existing ? existing.config_version + 1 : 1);
const { rows } = await getPool().query(
`INSERT INTO device_configs (device_id, config_version, settings_json, updated_at)
VALUES ($1,$2,$3,$4)
ON CONFLICT (device_id) DO UPDATE
SET config_version = EXCLUDED.config_version,
settings_json = EXCLUDED.settings_json,
updated_at = EXCLUDED.updated_at
RETURNING *`,
[payload.device_id, nextVersion, payload.settings_json, nowIso()]
);
return mapConfig(rows[0]);
}
export async function createDeviceConfigAck(payload: {
device_id: string;
config_version: number;
status: "applied" | "failed";
reason?: string;
payload_json?: Record<string, unknown>;
}): Promise<DeviceConfigAckEntity> {
const { rows } = await getPool().query(
`INSERT INTO device_config_acks (
id,
device_id,
config_version,
status,
reason,
payload_json,
acked_at
) VALUES ($1,$2,$3,$4,$5,$6,$7)
RETURNING *`,
[
`cfgack_${randomUUID()}`,
payload.device_id,
payload.config_version,
payload.status,
payload.reason || null,
payload.payload_json || {},
nowIso()
]
);
return mapAck(rows[0]);
}
export async function listDeviceConfigAcks(deviceId: string, limit = 50): Promise<DeviceConfigAckEntity[]> {
const { rows } = await getPool().query(
`SELECT * FROM device_config_acks
WHERE device_id = $1
ORDER BY acked_at DESC
LIMIT $2`,
[deviceId, Math.min(Math.max(limit, 1), 200)]
);
return rows.map(mapAck);
}
export function toDeviceConfigPayload(config: DeviceConfigEntity) {
return { ...config };
}
export function toDeviceConfigAckPayload(ack: DeviceConfigAckEntity) {
return { ...ack };
}

108
src/shared/store/ledgerStore.ts Normal file
View File

@ -0,0 +1,108 @@
import { randomUUID } from "node:crypto";
import { getPool } from "../db/pool";
import type { TransactionEntity } from "./transactionStore";
export interface LedgerEntryEntity {
id: string;
transaction_id: string;
merchant_id: string;
entry_type: "gross_income" | "platform_fee" | "merchant_payable";
amount: number;
currency: string;
direction: "credit" | "debit";
status: "posted" | "voided";
metadata_json: Record<string, unknown>;
created_at: string;
}
function nowIso() {
return new Date().toISOString();
}
function mapLedgerEntry(row: any): LedgerEntryEntity {
return {
id: row.id,
transaction_id: row.transaction_id,
merchant_id: row.merchant_id,
entry_type: row.entry_type,
amount: Number(row.amount),
currency: row.currency,
direction: row.direction,
status: row.status,
metadata_json: row.metadata_json || {},
created_at: row.created_at
};
}
export async function createPaidLedgerPlaceholder(tx: TransactionEntity): Promise<LedgerEntryEntity> {
const { rows } = await getPool().query(
`INSERT INTO ledger_entries (
id,
transaction_id,
merchant_id,
entry_type,
amount,
currency,
direction,
status,
metadata_json,
created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (transaction_id, entry_type) DO UPDATE
SET amount = EXCLUDED.amount,
currency = EXCLUDED.currency,
metadata_json = ledger_entries.metadata_json || EXCLUDED.metadata_json
RETURNING *`,
[
randomUUID(),
tx.id,
tx.merchant_id,
"gross_income",
tx.amount,
tx.currency,
"credit",
"posted",
{
placeholder: true,
source: "fase1_paid_transaction",
partner_reference: tx.partner_reference
},
nowIso()
]
);
return mapLedgerEntry(rows[0]);
}
export async function listLedgerEntries(filter?: {
transaction_id?: string;
merchant_id?: string;
limit?: number;
}): Promise<LedgerEntryEntity[]> {
const clauses: string[] = [];
const params: unknown[] = [];
let i = 1;
if (filter?.transaction_id) {
clauses.push(`transaction_id = $${i++}`);
params.push(filter.transaction_id);
}
if (filter?.merchant_id) {
clauses.push(`merchant_id = $${i++}`);
params.push(filter.merchant_id);
}
const limit = Math.min(Math.max(filter?.limit || 100, 1), 500);
const where = clauses.length ? `WHERE ${clauses.join(" AND ")}` : "";
const { rows } = await getPool().query(
`SELECT * FROM ledger_entries ${where} ORDER BY created_at DESC LIMIT ${limit}`,
params
);
return rows.map(mapLedgerEntry);
}
export function toLedgerEntryPayload(entry: LedgerEntryEntity) {
return { ...entry };
}

120
src/shared/store/mqttMessageStore.ts Normal file
View File

@ -0,0 +1,120 @@
import { randomUUID } from "node:crypto";
import { getPool } from "../db/pool";
export interface MqttMessageEntity {
id: string;
direction: "uplink" | "downlink";
device_id: string;
topic: string;
message_type: string;
correlation_id?: string;
payload_json: Record<string, unknown>;
publish_status: "recorded" | "sent" | "failed";
reason?: string;
created_at: string;
}
function nowIso() {
return new Date().toISOString();
}
function mapMessage(row: any): MqttMessageEntity {
return {
id: row.id,
direction: row.direction,
device_id: row.device_id,
topic: row.topic,
message_type: row.message_type,
correlation_id: row.correlation_id || undefined,
payload_json: row.payload_json || {},
publish_status: row.publish_status,
reason: row.reason || undefined,
created_at: row.created_at
};
}
export async function createMqttMessage(payload: {
direction: MqttMessageEntity["direction"];
device_id: string;
topic: string;
message_type: string;
correlation_id?: string;
payload_json?: Record<string, unknown>;
publish_status?: MqttMessageEntity["publish_status"];
reason?: string;
}): Promise<MqttMessageEntity> {
const { rows } = await getPool().query(
`INSERT INTO mqtt_messages (
id,
direction,
device_id,
topic,
message_type,
correlation_id,
payload_json,
publish_status,
reason,
created_at
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
RETURNING *`,
[
`mqtt_${randomUUID()}`,
payload.direction,
payload.device_id,
payload.topic,
payload.message_type,
payload.correlation_id || null,
payload.payload_json || {},
payload.publish_status || "recorded",
payload.reason || null,
nowIso()
]
);
return mapMessage(rows[0]);
}
export async function listMqttMessages(filter?: {
device_id?: string;
direction?: MqttMessageEntity["direction"];
message_type?: string;
correlation_id?: string;
limit?: number;
}): Promise<MqttMessageEntity[]> {
const clauses: string[] = [];
const params: unknown[] = [];
let i = 1;
if (filter?.device_id) {
clauses.push(`device_id = $${i++}`);
params.push(filter.device_id);
}
if (filter?.direction) {
clauses.push(`direction = $${i++}`);
params.push(filter.direction);
}
if (filter?.message_type) {
clauses.push(`message_type = $${i++}`);
params.push(filter.message_type);
}
if (filter?.correlation_id) {
clauses.push(`correlation_id = $${i++}`);
params.push(filter.correlation_id);
}
const limit = Math.min(Math.max(filter?.limit || 100, 1), 500);
const where = clauses.length ? `WHERE ${clauses.join(" AND ")}` : "";
const { rows } = await getPool().query(
`SELECT * FROM mqtt_messages ${where} ORDER BY created_at DESC LIMIT ${limit}`,
params
);
return rows.map(mapMessage);
}
export function toMqttMessagePayload(message: MqttMessageEntity) {
return { ...message };
}

13
src/shared/store/transactionStore.ts
View File

@ -1,5 +1,6 @@
import { randomUUID } from "node:crypto";
import { getPool } from "../db/pool";
import { createPaidLedgerPlaceholder } from "./ledgerStore";
export type TransactionStatus =
| "initiated"
@ -15,9 +16,10 @@ export type TransactionEventType =
| "CALLBACK_RECEIVED"
| "CALLBACK_REJECTED"
| "CALLBACK_DUPLICATE"
| "PUSH_QUEUED";
| "PUSH_QUEUED"
| "DYNAMIC_QR_CREATED";
export type TransactionEventSource = "webhook" | "system" | "admin";
export type TransactionEventSource = "webhook" | "system" | "admin" | "device";
export interface TransactionEntity {
id: string;
@ -265,7 +267,12 @@ export async function updateTransactionStatus(
}
});
return mapTransaction(rows[0]);
const updated = mapTransaction(rows[0]);
if (to === "paid") {
await createPaidLedgerPlaceholder(updated);
}
return updated;
}
export async function getTransactionById(id: string): Promise<TransactionEntity | null> {