diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 896cdd1..4fa8ff0 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -26,7 +26,7 @@ jobs: uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: repository: QRun-IO/qqq-print-protocol - ref: cc3c4e50e8f6f4fbc93d351fb51edb4f5438bbd3 + ref: 67be986cb76d6b5d9a9f72cad054193b0d11b0d5 path: qqq-print-protocol persist-credentials: false - uses: actions/setup-java@be666c2fcd27ec809703dec50e508c2fdc7f6654 # v5.2.0 diff --git a/CHANGELOG.md b/CHANGELOG.md index c38cf48..fb09571 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## 0.3.0-rc.1 — unreleased + +- Add client-initiated realtime WebSocket delivery under native printing API 1.1, + with current QQQ session and record permissions revalidated during connection use. +- Deliver desired state, device commands, inventory, lifecycle events and receipts + over the socket; native claims require acknowledged current settings and state. +- Stream retained documents independently from controls and expose document, + concurrency, timing, admission and retention batch limits in native admin forms. +- Preserve durable job/command identities across reconnects. Deployment and actual + Windows realtime acceptance remain pending until the candidate gates complete. + ## 0.2.0-rc.1 — 2026-10-31 - Add native device/table selection actions for pause, resume, disconnect, diff --git a/README.md b/README.md index a079476..ef69717 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # QQQ Printing -Cloud printing for QQQ 4.1+, paired with the [Windows AppKit connector](https://github.com/QRun-IO/qqq-print-agent) and the [shared protocol](https://github.com/QRun-IO/qqq-print-protocol). The Windows connector initiates outbound HTTPS polling; the cloud needs no inbound route into the facility. +Cloud printing for QQQ 4.1+, paired with the [Windows AppKit connector](https://github.com/QRun-IO/qqq-print-agent) and the [shared protocol](https://github.com/QRun-IO/qqq-print-protocol). The Windows connector opens an outbound, authenticated two-way WebSocket; documents download separately over HTTPS. The cloud needs no inbound route into the facility. The real-time transport described here is unreleased source; published 0.2.0-rc.1 uses the earlier polling transport. **Prerelease software**, tested against QQQ **4.1.0-RC.20** and Java 21. The current published release is [**0.2.0-rc.1**](https://github.com/QRun-IO/qbit-printing/releases/tag/v0.2.0-rc.1). The tagged build passed 110 tests and zero Checkstyle violations. The optional native UI demo uses QQQ frontend **1.0.0-RC.37**. GitHub releases provide a library JAR, POM and SHA-256 checksums, with verified build attestations for all three assets. This repository does not claim Maven Central publication or provide a standalone executable/fat JAR. @@ -11,16 +11,17 @@ The 0.2.0-rc.1 source adds native record/selection actions, per-device printers/ ```mermaid flowchart LR A[QQQ application / durable job] --> B[Printing QBit and JDBC queue] - C[Windows service] -->|Outbound HTTPS polling| B + C[Windows service] <-->|Client-initiated WSS: commands and status| B + C -->|Separate outbound HTTPS document streams| B C --> D[Windows spooler] D --> E[Local laser / Zebra] ``` -See the [screen walkthrough](docs/UI-WALKTHROUGH.md), [storage and retention](docs/STORAGE-AND-RETENTION.md), [operations and troubleshooting](docs/OPERATIONS.md), [contributing](CONTRIBUTING.md), [security reporting](SECURITY.md), [changelog](CHANGELOG.md), and [Apache-2.0 license](LICENSE). +See the [real-time host integration](docs/REALTIME.md), [screen walkthrough](docs/UI-WALKTHROUGH.md), [storage and retention](docs/STORAGE-AND-RETENTION.md), [operations and troubleshooting](docs/OPERATIONS.md), [contributing](CONTRIBUTING.md), [security reporting](SECURITY.md), [changelog](CHANGELOG.md), and [Apache-2.0 license](LICENSE). ## Build and verify -Install Java 21. Clone this repository and `qqq-print-protocol` as siblings. This source pins protocol commit `cc3c4e50e8f6f4fbc93d351fb51edb4f5438bbd3`; use that commit for reproducible compatibility tests. Use the checked-in wrapper (Maven 3.9.11). +Install Java 21. Clone this repository and `qqq-print-protocol` as siblings. This source pins protocol commit `67be986cb76d6b5d9a9f72cad054193b0d11b0d5`; use that commit for reproducible compatibility tests. Use the checked-in wrapper (Maven 3.9.11). ```powershell .\mvnw.cmd verify @@ -120,7 +121,7 @@ Metadata names follow QBit conventions: tables and data sources use `printing_*` Two independent settings govern dispatch: `available` means the queue is present in the latest locally selected inventory; `enabled` means a QQQ administrator approved it. New submissions and claims require both. Removing a queue from local selection makes it unavailable without deleting its approval/history. An empty inventory withdraws all queues. Withdrawing a queue cannot retract a job already accepted by Windows. -Application code calls `printing.submit(JSONObject)` with the shared `SubmitJobRequest` fields. A stable business idempotency key returns the same job when retried; changing document/options/target under the same key fails. PDF is sent as bytes encoded once in base64; ZPL is UTF-8 printer-ready bytes. Default limit is 20 MiB. Queue approval and format compatibility are checked before insertion. +Application code calls `printing.submit(JSONObject)` with the shared `SubmitJobRequest` fields. A stable business idempotency key returns the same job when retried; changing document/options/target under the same key fails. PDF is sent as bytes encoded once in base64; ZPL is UTF-8 printer-ready bytes. Default decoded limits are 100 MiB for PDF and 20 MiB for ZPL, both configurable in the native Printing connection and limits process. Queue approval and format compatibility are checked before insertion. ```java JSONObject job = printing.submit(new JSONObject() diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md index 1b57c0a..02d2f15 100644 --- a/docs/OPERATIONS.md +++ b/docs/OPERATIONS.md @@ -40,7 +40,7 @@ The host configures resource ceilings through `PrintingQBitConfig.withAdmissionL `PrintingAdmissionLimits.defaults()` returns these defaults. To override them, construct the record with device/tenant pairs in this order: outstanding jobs, outstanding bytes, retained bytes, hourly admissions, cumulative printers. Values must be positive. Use the same configuration on every QQQ node serving a tenant. Size retained capacity for actual document sizes and the chosen retention period; 10,000 jobs/day is a sizing target, not a measured hardware throughput result. These are logical application byte limits, not an S3 billing limit: object versions, multipart fragments and provider overhead require their own deployment monitoring and lifecycle policy. -Admission is serialized per tenant in SQL before storage writes. Pending and failed upload reservations stay charged until exact-object cleanup succeeds; failed admissions that reached a committed reservation also consume the hourly allowance. Legacy reservations with unknown length conservatively reserve the maximum 20 MiB each, against the tenant when their device is unknown. Existing payloads are backfilled without removal. Apply this additive accounting migration with older application writers stopped; mixed-version writers or direct SQL changes bypass accounting and are unsupported. +Admission is serialized per tenant in SQL before storage writes. Pending and failed upload reservations stay charged until exact-object cleanup succeeds; failed admissions that reached a committed reservation also consume the hourly allowance. Legacy reservations with unknown length block new allocations until exact cleanup resolves them; a fixed-size guess cannot safely account for configurable document limits. Existing payloads are backfilled without removal. Apply this additive accounting migration with older application writers stopped; mixed-version writers or direct SQL changes bypass accounting and are unsupported. HTTP 429 means capacity, the rolling-hour allowance, cumulative inventory history, or an existing upload reservation prevents a new allocation. The response explains the category. Reuse the same idempotency/request identity when retrying: a completed matching replay returns its existing job even at capacity or after the original retry source expires. Wait for outstanding work or the hourly window, investigate failed cleanup/storage access, or have the host administrator adjust a justified capacity limit. Retention only removes eligible successful documents; it does not erase failed, uncertain or active work to make space. @@ -125,13 +125,12 @@ Only a legacy-only host such as `PrintingDemo` should opt into `PrintingRoutes.LegacyEnrollment.ENABLED`. A rejected legacy enrollment does not consume the code, which still requires native user and facility authorization. -Pause blocks new claims. Disconnect also expresses that ordinary agent heartbeat, -inventory publication, and claiming should stop; the independent authenticated -control poll remains available for reconnect. Neither operation cancels in-flight -jobs. Resume preserves connection state; reconnect preserves pause. Revocation -invalidates the enrolled credential permanently, blocks control polling, and -requires separate local re-enrollment. It cannot be undone with resume/reconnect. -Desired state, the last observed revision, and the last poll are separate facts: +Pause blocks new claims. Disconnect stops dispatch and inventory publication while +the authenticated control WebSocket remains available for a reconnect command. +Neither operation cancels an already submitted spooler job. Resume preserves +connection state; reconnect preserves pause. Revocation invalidates device proof, +closes its channel and requires separate local re-enrollment. It cannot be undone +with resume/reconnect. Desired state, observed revision and last contact are separate facts: a successful server action does not prove an offline device applied it. Queue clear atomically changes only this device's QUEUED jobs to FAILED with the @@ -171,7 +170,7 @@ cancellation request, not that paper was withdrawn or a physical print was undon | --- | --- | | Enrollment rejected | Code expiration/single use, correct server, administrator-created facility/device enrollment | | No printers in QQQ | Service access to Windows queues; local exposure selection; connector enrollment and inventory publication | -| Printer listed but submission rejected | Both `available` and `enabled`, format capability and 20 MiB decoded-size limit | +| Printer listed but submission rejected | Both `available` and `enabled`, format capability and configured PDF/ZPL decoded-size limits | | Job stays queued | Connector connectivity, selected queue availability and device assignment | | Job stays received/submitting after restart | Inspect local journal/spooler; automatic uncertain-output replay is intentionally absent | | Printing app missing or access denied | QBit registration, `printing_admin`, matching `printing_tenant`, and correct QQQ backend | diff --git a/docs/REALTIME.md b/docs/REALTIME.md new file mode 100644 index 0000000..e94dd41 --- /dev/null +++ b/docs/REALTIME.md @@ -0,0 +1,100 @@ +# Real-time printing host integration + +This source implements printing extension **1.1.0** (wire envelope v1). It is not +part of the published 0.2.0-rc.1 artifacts. Upgrade server and Windows connector +together; a real-time connector rejects a server that advertises only extension +1.0. There is no timed job/control polling fallback. + +## Connection and authorization + +The Windows service initiates `wss:///qqq/v1/qbit/printing/device/realtime` +using subprotocol `qqq-printing.v1`. No facility listener or inbound firewall rule +is required. Plain WebSocket is allowed only for explicit loopback development. +Configure the cloud ingress to forward WebSocket upgrades, preserve the original +HTTPS origin and allow the configured heartbeat interval/timeout. + +The service discovers authentication through the QQQ UI/UX API and uses the +standard current QQQ session, client channel and approved bound-device proof. +Credentials are not URL parameters. Device proof does not replace the user's +authority. The server validates the current session, user/tenant, channel policy +and operation permissions; socket establishment does not grant permanent rights. +Session expiry, logout, revocation and permission denial require correction or +sign-in through the configured QQQ provider. Reconnect cannot bypass them. + +Register the route provider alongside the ordinary native QQQ endpoints: + +```java +server.withAdditionalRouteProviders(List.of(existingHttpRoutes, + new PrintingRealtimeRoutes(printingService))); +``` + +Use the **same PrintingService instance** in the QBit configuration and route +provider. `PrintingLocalHost.configure` shows the complete launcher integration. +The native HTTP claim route requires a connected socket that has applied the +current desired state and settings. Registering just the older HTTP provider is +insufficient. Close the provider when its host stops. + +## Independent control and document lanes + +The socket carries small JSON state, commands, inventory, heartbeats, job events +and durable receipts. A job notification triggers a finite HTTP claim drain; +an empty queue waits for another push. No timer repeatedly fetches claims or +commands. Heartbeats, session renewal and reconnect backoff are lifecycle work, +not job polling. + +Documents use separate outbound same-origin HTTPS transfers. The Windows service +streams them to protected staging files and checks declared length and SHA-256 +before native submission. The socket reader remains active during a slow download. +PDF rendering and ZPL spooling receive the verified file rather than a base64 +payload in the control pipe. Limits bound staging, messages and concurrency. + +Receipts remain on disk until the server acknowledges their committed IDs. A +reconnect resynchronizes current state/settings and replays retained receipts. +An interrupted native submission remains uncertain and is never automatically +reprinted. A successful spooler submission does not prove paper physically printed. + +## Native administrator settings + +Printing home links to two standard QQQ processes. They enforce current +administrator and tenant access on the backend as well as in metadata: + +- **Printing connection and limits**: PDF/ZPL size, copies, staging space, + concurrent downloads/prints, bandwidth, request/download timeouts, heartbeat, + reconnect backoff/jitter, acknowledgment timeout and control message queues. + Defaults are 100 MiB PDF, 20 MiB ZPL and one download/print lane. Saved policy + changes have a durable revision and are pushed to connected devices. +- **Printing queue and storage limits**: device/tenant outstanding counts and + bytes, retained bytes, hourly allocations and printer counts. New admissions + use the persisted policy inside their SQL transaction. Reducing limits preserves + existing jobs and history; it does not silently remove documents. + +Fields use readable MiB/GiB/seconds with exact conversion, required values and +cross-field validation. Unsupported client ceilings fail explicitly rather than +silently clamping. Storage retention remains in the separate native retention +process, with a one-year default. + +HTTP ingress/body limits remain host-owned. The local host derives its upload +body ceiling at startup from the persisted document limits (including base64 +expansion for legacy submission). Raising a document limit above the running +transport ceiling requires a host restart or corresponding production ingress +configuration. Download streaming does not base64-expand documents. + +## Host ownership and deployment boundary + +Commit wakeups are emitted after SQL commit. A subscription exists before initial +state reconstruction, closing the empty-claim/reconnect notification race. +`PrintingRealtimeBus` is an explicit host-owned notification and connection +ownership seam. The default implementation is in-process; this source does **not** +provide a distributed broker or claim tested multi-node production support. + +Use one writer/application host with its configured service/bus. Multiple route +providers sharing that explicit in-process bus fence a device's prior connection. +A multi-node deployment needs a host-provided shared implementation preserving +both post-commit notifications and single-channel ownership, and its own +qualification. Separate independent service instances do not magically publish +notifications to one another. + +The existing `/printing/v1` routes remain explicit legacy diagnostic compatibility. +They are not a real-time fallback and cannot authorize a native bound device. +The full message and settings contract is in +[the shared protocol](https://github.com/QRun-IO/qqq-print-protocol). diff --git a/docs/SESSION-STATE.md b/docs/SESSION-STATE.md index 379a218..7ac9ea0 100644 --- a/docs/SESSION-STATE.md +++ b/docs/SESSION-STATE.md @@ -82,3 +82,13 @@ Final QBit UI pass: metadata uses printing_* table/data-source names, printing-p Durable target correction: core DurableJobsRuntime.effectLoop constructs a system session without user tenant keys. PrintEnqueueEffectTarget now has a private trusted resolver for that system-only outbox path, bound to the current instance PrintingQBitConfig service and its fixed tenant. Inline users retain scope checks; native process authorization is unchanged. Registered-target tests cover system deliver/reconcile, idempotent replay, foreign-tenant device rejection, cached target reuse across instances, and inline wrong/missing-tenant denial. `mvnw.cmd -o verify -Pui-demo dependency:build-classpath` passed all 16 tests and zero Checkstyle violations after the durable target correction, explicit disabled table writes, final dashboard timestamps, and frontend RC.37 profile update. UI runtime classpath regenerated at target/ui-runtime-classpath.txt. Primary identifiers remain non-hidden because QQQ excludes hidden fields from ordinary query results; explicit frontend column hints keep them out of the default operational grid. + +## Real-time transport implementation resumed — 2026-10-31 + +User resumed implementation after explicitly rejecting periodic job/control polling. Active isolated worktrees: qbit-printing-wt-realtime, qqq-print-protocol-wt-realtime, qqq-print-agent-wt-job-table (all branch feat/realtime-printing-20261031). Installed 0.2.0-rc.1 remains unchanged and is still polling; do not claim it is real-time. + +Protocol extension 1.1.0: client-initiated authenticated WSS, qqq-printing.v1 subprotocol, current QQQ session and secondary bound-device proof. Small state/commands/events/receipts travel directly over WebSocket; separate HTTPS documents stream to protected files. No timed polling fallback. Commit notifications and finite event-triggered claims, bounded concurrency, persistent receipts and reconnect reconciliation are being implemented. Tenant administrator forms expose document, transfer, connection, queue/rate/printer/storage limits. Default PDF limit100MiB, ZPL20MiB; both configurable. Root owns native forms/admission settings, server agent owns route/storage/service, Windows agent owns transport/pipeline, inventory agent completed protocol70definitions/72fixtures and now owns Jobs UI expansion/sorting. No live install or new physical print yet. New-source gates/integration are in progress; previous0.2 gates do not qualify these changes. + +## Real-time source verification — 2026-10-31 + +Server source0.3.0-rc.1 passes117 Java tests and zero Checkstyle violations, including real native-session WebSocket lifecycle, sessionId/sessionUUID, logout/channel rejection, replacement fencing, partial revision acknowledgments, no idle-inventory feedback, and21MiB native storage streaming/integrity. Protocol70definitions/72fixtures passes; both consumer workflows pin signed protocol67be986cb76d6b5d9a9f72cad054193b0d11b0d5. Native admin policy tests cover permission/tenant checks, exact250.5MiB conversion, revision persistence and invalid-capacity rejection. A private actual.NET/Java fixture passed13 functional checks with25MiB content, durable receipt replay, channel replacement and sign-out denial; its initial fixture JVM needed explicit maintenance-thread shutdown handling, so this is not installed lifecycle qualification. Root is preparing the final refreshed fixture run. Current installed0.2 remains unchanged; release/install acceptance is pending. diff --git a/docs/TODO.md b/docs/TODO.md index 099ed8a..bbdc046 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -79,3 +79,11 @@ The Windows gate, MSI rebuild, installed UI acceptance, release evidence and fre - [x] Complete independent review after the five baseline security findings are remediated; current details are in SESSION-STATE.md. - [ ] Complete remote deployment and production S3/IAM qualification; local native acceptance and isolated storage tests do not establish these. + +## Real-time upgrade (active) + +- [ ] Complete authenticated outbound WebSocket transport and automatic durable recovery; remove production job/control polling fallback. +- [ ] Verify concurrent control while large streamed documents download, bounds/cleanup, real concurrency and exact event replay. +- [ ] Verify native administrator limits forms, live policy delivery, admission overrides and configurable document sizes. +- [ ] Complete consistent sortable/expandable Windows Jobs and Printers tables. +- [ ] Run coordinated server/client/protocol gates and cross-runtime native integration before packaging/installing. diff --git a/pom.xml b/pom.xml index a2b161c..4d3bbcf 100644 --- a/pom.xml +++ b/pom.xml @@ -4,7 +4,7 @@ com.kingsrookqbit-build-parent2.0.0 com.kingsrook.qbits qbit-printing - 0.2.0-rc.1 + 0.3.0-rc.1 QQQ Printing Apache License, Version 2.0https://www.apache.org/licenses/LICENSE-2.0 diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingAdmission.java b/src/main/java/com/kingsrook/qbits/printing/PrintingAdmission.java index c81bf9b..b1bb55d 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingAdmission.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingAdmission.java @@ -153,6 +153,10 @@ final class PrintingAdmission ***************************************************************************/ static void reserve(Connection connection, String tenant, String device, String id, int bytes, PrintingAdmissionLimits limits) throws SQLException { + if(number(connection, "SELECT COUNT(*) FROM printing_document_orphan WHERE tenant_id=? AND content_length IS NULL", tenant) != 0) + { + throw new PrintingException(429, "An old upload has an unknown document size; complete exact orphan cleanup before admitting new documents"); + } if(number(connection, "SELECT COUNT(*) FROM printing_resource_usage WHERE tenant_id=? AND device_id=?", tenant, device) == 0) { execute(connection, "INSERT INTO printing_resource_usage VALUES (?, ?, 0)", tenant, device); diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingAdmissionSettings.java b/src/main/java/com/kingsrook/qbits/printing/PrintingAdmissionSettings.java new file mode 100644 index 0000000..254ce07 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingAdmissionSettings.java @@ -0,0 +1,178 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import javax.sql.DataSource; +import java.math.BigDecimal; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.List; +import java.util.Objects; +import org.json.JSONObject; + + +/******************************************************************************* + ** Tenant admission overrides are read in the same transaction as admission. + *******************************************************************************/ +public class PrintingAdmissionSettings +{ + private static final List FIELDS = List.of("deviceOutstandingJobs", "tenantOutstandingJobs", "deviceOutstandingBytes", "tenantOutstandingBytes", + "deviceRetainedBytes", "tenantRetainedBytes", "deviceJobsPerHour", "tenantJobsPerHour", "devicePrinters", "tenantPrinters"); + private static final long MAX_EXACT_JSON_INTEGER = 9_007_199_254_740_991L; + private final DataSource source; + private final String tenant; + private final PrintingAdmissionLimits defaults; + + + + /*************************************************************************** + ** The host supplies immutable defaults until an administrator saves an override. + ***************************************************************************/ + public PrintingAdmissionSettings(DataSource source, String tenant, PrintingAdmissionLimits defaults) + { + this.source = Objects.requireNonNull(source); + this.tenant = Objects.requireNonNull(tenant); + this.defaults = Objects.requireNonNull(defaults); + } + + + + /*************************************************************************** + ** The enclosing service ensures the tenant settings row before public reads. + ***************************************************************************/ + public JSONObject get() throws SQLException + { + try(Connection connection = source.getConnection()) + { + return (json(effective(connection))); + } + } + + + + /*************************************************************************** + ** Validate all limits before one atomic tenant-scoped update. + ***************************************************************************/ + public JSONObject save(JSONObject requested) throws SQLException + { + PrintingAdmissionLimits limits = parse(requested); + JSONObject value = json(limits); + try(Connection connection = source.getConnection(); var update = connection.prepareStatement("UPDATE printing_settings SET admission_policy_json=? WHERE tenant_id=?")) + { + update.setString(1, value.toString()); + update.setString(2, tenant); + if(update.executeUpdate() != 1) + { + throw new SQLException("The tenant admission settings row is unavailable"); + } + } + return (value); + } + + + + /*************************************************************************** + ** Do not open a nested connection while a queue transaction holds its locks. + ***************************************************************************/ + public PrintingAdmissionLimits effective(Connection connection) throws SQLException + { + try(var query = connection.prepareStatement("SELECT admission_policy_json FROM printing_settings WHERE tenant_id=?")) + { + query.setString(1, tenant); + try(var rows = query.executeQuery()) + { + if(!rows.next()) + { + return (defaults); + } + String stored = rows.getString(1); + return (stored == null ? defaults : parse(new JSONObject(stored))); + } + } + } + + + + /*************************************************************************** + ** Every configured limit is explicit, positive and exactly representable. + ***************************************************************************/ + private static PrintingAdmissionLimits parse(JSONObject value) + { + if(!value.keySet().equals(java.util.Set.copyOf(FIELDS))) + { + throw new IllegalArgumentException("Specify all admission limits without unknown fields"); + } + try + { + return (new PrintingAdmissionLimits(integer(value, "deviceOutstandingJobs"), integer(value, "tenantOutstandingJobs"), + number(value, "deviceOutstandingBytes"), number(value, "tenantOutstandingBytes"), number(value, "deviceRetainedBytes"), number(value, "tenantRetainedBytes"), + integer(value, "deviceJobsPerHour"), integer(value, "tenantJobsPerHour"), integer(value, "devicePrinters"), integer(value, "tenantPrinters"))); + } + catch(ArithmeticException exception) + { + throw new IllegalArgumentException("Admission counts must fit the supported integer range", exception); + } + } + + + + /*************************************************************************** + ** Counts follow the database's integer representation without truncation. + ***************************************************************************/ + private static int integer(JSONObject value, String field) + { + return (Math.toIntExact(number(value, field))); + } + + + + /*************************************************************************** + ** Reject quoted numbers, fractions, overflow and disabled capacity boundaries. + ***************************************************************************/ + private static long number(JSONObject value, String field) + { + if(!(value.opt(field) instanceof Number number)) + { + throw new IllegalArgumentException(field + " must be a positive integer"); + } + long parsed = new BigDecimal(number.toString()).longValueExact(); + if(parsed < 1 || parsed > MAX_EXACT_JSON_INTEGER) + { + throw new IllegalArgumentException(field + " is outside the supported positive range"); + } + return (parsed); + } + + + + /*************************************************************************** + ** Native forms and host defaults share one stable field vocabulary. + ***************************************************************************/ + private static JSONObject json(PrintingAdmissionLimits value) + { + return (new JSONObject().put("deviceOutstandingJobs", value.deviceOutstandingJobs()).put("tenantOutstandingJobs", value.tenantOutstandingJobs()) + .put("deviceOutstandingBytes", value.deviceOutstandingBytes()).put("tenantOutstandingBytes", value.tenantOutstandingBytes()) + .put("deviceRetainedBytes", value.deviceRetainedBytes()).put("tenantRetainedBytes", value.tenantRetainedBytes()) + .put("deviceJobsPerHour", value.deviceJobsPerHour()).put("tenantJobsPerHour", value.tenantJobsPerHour()) + .put("devicePrinters", value.devicePrinters()).put("tenantPrinters", value.tenantPrinters())); + } +} diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingDeviceControls.java b/src/main/java/com/kingsrook/qbits/printing/PrintingDeviceControls.java index 741b501..beb3cf2 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingDeviceControls.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingDeviceControls.java @@ -281,6 +281,7 @@ final class PrintingDeviceControls { throw new PrintingException(409, "Unknown control revision"); } + applied = Math.max(applied, device.getLong("control_applied_revision")); String now = Instant.now().toString(); update(connection, "UPDATE printing_device SET control_version=1,control_capabilities=?,control_last_poll_at=?,control_applied_revision=?,agent_version=?,last_seen_at=? WHERE tenant_id=? AND id=?", capabilities.toString(), now, applied, version, now, tenant, deviceId); diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStorage.java b/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStorage.java index de7f8a9..7445564 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStorage.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStorage.java @@ -94,15 +94,22 @@ public class PrintingDocumentStorage ** Bound reads even if stored content was replaced outside this service. ***************************************************************************/ public byte[] get(String table, String reference) throws SQLException + { + return (get(table, reference, Math.toIntExact(PrintingRealtimePolicy.defaults().get("maxPdfBytes")))); + } + + + /** Internal retry callers supply the exact persisted length, independent of the current format policy. */ + public byte[] get(String table, String reference, int maximum) throws SQLException { return (execute(() -> { try(InputStream stream = new StorageAction().getInputStream(input(table, reference))) { - byte[] bytes = stream.readNBytes(20 * 1024 * 1024 + 1); - if(bytes.length > 20 * 1024 * 1024) + byte[] bytes = stream.readNBytes(maximum); + if(stream.read() != -1) { - throw new PrintingException(502, "Stored print document exceeds the allowed size"); + throw new PrintingException(502, "Stored print document exceeds its declared length"); } return bytes; } @@ -110,6 +117,13 @@ public class PrintingDocumentStorage } + /** Return the native backend stream; the request owns closing it and enforcing its declared length. */ + public InputStream open(String table, String reference) throws SQLException + { + return (execute(() -> new StorageAction().getInputStream(input(table, reference)))); + } + + /*************************************************************************** ** Backend deletion is idempotent; callers mark receipts only after success. diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStream.java b/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStream.java new file mode 100644 index 0000000..91e6d06 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingDocumentStream.java @@ -0,0 +1,90 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.io.FilterInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; + + +/** Length and digest verification remains bounded while transferring a retained object. */ +final class PrintingDocumentStream extends FilterInputStream +{ + private final long length; + private final String expectedHash; + private final MessageDigest digest; + private long received; + private boolean verified; + + + /** Wrap a source with immutable document integrity metadata. */ + PrintingDocumentStream(InputStream input, long length, String expectedHash) + { + super(input); + this.length = length; + this.expectedHash = expectedHash; + try + { + digest = MessageDigest.getInstance("SHA-256"); + } + catch(NoSuchAlgorithmException exception) + { + throw new IllegalStateException(exception); + } + } + + + @Override + public int read() throws IOException + { + byte[] one = new byte[1]; + return (read(one, 0, 1) == -1 ? -1 : Byte.toUnsignedInt(one[0])); + } + + + @Override + public int read(byte[] bytes, int offset, int count) throws IOException + { + int read = in.read(bytes, offset, (int) Math.min(count, Math.max(1, length - received + 1))); + if(read > 0) + { + received += read; + if(received > length) + { + throw new IOException("Stored print document exceeds its declared length"); + } + digest.update(bytes, offset, read); + } + else if(read == -1 && !verified) + { + verified = true; + if(received != length || !HexFormat.of().formatHex(digest.digest()).equals(expectedHash)) + { + throw new IOException("Stored print document integrity check failed"); + } + } + return (read); + } +} \ No newline at end of file diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingHome.java b/src/main/java/com/kingsrook/qbits/printing/PrintingHome.java index 6425b7a..b127dac 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingHome.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingHome.java @@ -153,6 +153,8 @@ public class PrintingHome extends AbstractWidgetRenderer .withBlock(text("Scope: current tenant. Devices/printers: now. Job exceptions/recent work: last 24 hours.")) .withBlock(text("Enroll a print device").withLink(new BlockLink("/app/printingEnrollDevice/?returnTo=%2Fapp%2Fprinting%2F"))) .withBlock(text("Printing retention settings").withLink(new BlockLink("/app/printingRetentionSettings/?returnTo=%2Fapp%2Fprinting%2F"))) + .withBlock(text("Printing connection and limits").withLink(new BlockLink("/app/printingRealtimeSettings/?returnTo=%2Fapp%2Fprinting%2F"))) + .withBlock(text("Printing queue and storage limits").withLink(new BlockLink("/app/printingAdmissionSettings/?returnTo=%2Fapp%2Fprinting%2F"))) .withBlock(text("Run retention cleanup").withLink(new BlockLink("/app/printingRunRetention/?returnTo=%2Fapp%2Fprinting%2F"))) .withBlock(text("Printing setup and help").withLink(new BlockLink("https://github.com/QRun-IO/qbit-printing/blob/main/docs/OPERATIONS.md"))); if(count(DEVICE, new QQueryFilter()) == 0) diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingLocalHost.java b/src/main/java/com/kingsrook/qbits/printing/PrintingLocalHost.java index 44dbd46..b90eafc 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingLocalHost.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingLocalHost.java @@ -132,13 +132,23 @@ public class PrintingLocalHost extends AbstractQQQApplication ***************************************************************************/ void configure(QApplicationJavalinServer server, boolean frontend) { + long uploadBodyLimit; + try + { + PrintingRealtimePolicy policy = service.realtimePolicy(); + uploadBodyLimit = ((Math.max(policy.get("maxPdfBytes"), policy.get("maxZplBytes")) + 2) / 3) * 4 + 65536; + } + catch(java.sql.SQLException exception) + { + throw new IllegalStateException("Cannot load printing transport limits", exception); + } PrintingRoutes routes = new PrintingRoutes(service, context -> MessageDigest.isEqual( settings.administratorToken().getBytes(StandardCharsets.UTF_8), PrintingRoutes.bearer(context).getBytes(StandardCharsets.UTF_8)), PrintingRoutes.LegacyEnrollment.DISABLED); server.withPort(settings.port()).withServeFrontendNext(frontend).withServeFrontendMaterialDashboard(false) - .withAdditionalRouteProviders(List.of(routes)).withJavalinConfigCustomizer(config -> + .withAdditionalRouteProviders(List.of(routes, new PrintingRealtimeRoutes(service))).withJavalinConfigCustomizer(config -> { config.jetty.host = "127.0.0.1"; - config.http.maxRequestSize = 29 * 1024 * 1024; + config.http.maxRequestSize = uploadBodyLimit; config.routes.before(PrintingLocalHost::guardOrigin); }); } diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java b/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java index ca31f1f..053e8a8 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingQBitProducer.java @@ -176,6 +176,7 @@ public class PrintingQBitProducer implements QBitProducer PrintingDeviceProcesses.register(instance); PrintingUiApi.register(instance); PrintingRetentionProcesses.register(instance, config); + PrintingRealtimeProcesses.register(instance, app); PrintingHome.register(instance, app); instance.addApp(app); QAppMetaData parent = instance.getApp("admin"); diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java new file mode 100644 index 0000000..5662ec8 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeBus.java @@ -0,0 +1,91 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.BiConsumer; + + +/** Host-owned commit notification bridge. Callbacks only schedule bounded work. */ +public class PrintingRealtimeBus +{ + private final CopyOnWriteArrayList> listeners = new CopyOnWriteArrayList<>(); + private final java.util.concurrent.ConcurrentHashMap owners = new java.util.concurrent.ConcurrentHashMap<>(); + + + /** Atomically fence the preceding connection across route providers sharing this host bus. */ + public void acquire(String tenant, String device, String generation, Runnable replaced) + { + Owner previous = owners.put(tenant + ":" + device, new Owner(generation, replaced)); + if(previous != null) + { + previous.replaced().run(); + } + } + + + /** A replaced connection cannot claim or send even if its local close callback is delayed. */ + public boolean owns(String tenant, String device, String generation) + { + Owner current = owners.get(tenant + ":" + device); + return (current != null && current.generation().equals(generation)); + } + + + /** Release only the exact generation; old disconnects cannot remove a replacement. */ + public void release(String tenant, String device, String generation) + { + owners.computeIfPresent(tenant + ":" + device, (key, owner) -> owner.generation().equals(generation) ? null : owner); + } + + + /** Ownership contains no device token or user credential. */ + private record Owner(String generation, Runnable replaced) + { + } + + + /** Register before reading the durable snapshot to avoid a reconnect lost wakeup. */ + public AutoCloseable subscribe(BiConsumer listener) + { + listeners.add(listener); + return (() -> listeners.remove(listener)); + } + + + /** Publish only after commit. Shared-host implementations must preserve this nonblocking contract. */ + public void publish(String tenant, String device) + { + for(BiConsumer listener : listeners) + { + try + { + listener.accept(tenant, device); + } + catch(RuntimeException ignored) + { + // A stopped listener must not turn a successful commit into an apparent failure. + // Reconnection always reconstructs work from the durable database. + } + } + } +} diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimePolicy.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimePolicy.java new file mode 100644 index 0000000..5116fb6 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimePolicy.java @@ -0,0 +1,172 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.util.LinkedHashMap; +import java.util.Map; +import org.json.JSONObject; + + +/** Validated administrator policy, independent of any individual client connection. */ +public final class PrintingRealtimePolicy +{ + /** Canonical policy defaults and validation ranges. */ + private static final Map FIELDS = fields(); + private final JSONObject values; + + + /** Keep the validated policy private. */ + private PrintingRealtimePolicy(JSONObject values) + { + this.values = values; + } + + + /** Operational defaults; administrators can explicitly change every limit. */ + public static PrintingRealtimePolicy defaults() + { + JSONObject values = new JSONObject(); + FIELDS.forEach((name, range) -> values.put(name, range[0])); + return (new PrintingRealtimePolicy(values)); + } + + + /** Require a complete policy and reject unknown, fractional or out-of-range values. */ + public static PrintingRealtimePolicy fromJson(JSONObject input) + { + if(!input.keySet().equals(FIELDS.keySet())) + { + throw new PrintingException(400, "Realtime settings require exactly the documented policy fields"); + } + JSONObject values = new JSONObject(); + FIELDS.forEach((name, range) -> + { + Object raw = input.get(name); + if(!(raw instanceof Number number) || !number.toString().matches("[0-9]+")) + { + throw new PrintingException(400, "Realtime setting must be an integer: " + name); + } + long value; + try + { + value = Long.parseLong(number.toString()); + } + catch(NumberFormatException exception) + { + throw new PrintingException(400, "Realtime setting exceeds its range: " + name); + } + if(value < range[1] || value > range[2]) + { + throw new PrintingException(400, "Realtime setting exceeds its range: " + name); + } + values.put(name, value); + }); + PrintingRealtimePolicy policy = new PrintingRealtimePolicy(values); + policy.atLeast("maxQueuedFrameBytes", policy.get("maxFrameBytes")); + policy.atLeast("maxStagedBytes", Math.max(policy.get("maxPdfBytes"), policy.get("maxZplBytes"))); + policy.atLeast("downloadTimeoutMillis", Math.max(policy.get("downloadConnectTimeoutMillis"), policy.get("downloadIdleTimeoutMillis"))); + policy.atLeast("heartbeatTimeoutSeconds", 2 * policy.get("heartbeatIntervalSeconds")); + policy.atLeast("reconnectMaxDelayMillis", policy.get("reconnectMinDelayMillis")); + return (policy); + } + + + /** Read a validated named setting without exposing mutable JSON. */ + public long get(String name) + { + return (values.getLong(name)); + } + + + /** Wire settings include a server-generated durable revision. */ + public JSONObject toJson(long revision) + { + return (new JSONObject(values.toString()).put("revision", revision)); + } + + + /** Submission and retained-document retrieval use the configured format bound. */ + public long documentLimit(String format) + { + if(!"PDF".equals(format) && !"ZPL".equals(format)) + { + throw new PrintingException(400, "Only PDF and ZPL are supported"); + } + return (get("PDF".equals(format) ? "maxPdfBytes" : "maxZplBytes")); + } + + + /** A client must explicitly support the policy; never silently lower administrator limits. */ + public void requireClientLimits(JSONObject limits) + { + for(String name : new String[] {"maxPdfBytes", "maxZplBytes", "maxFrameBytes", "maxQueuedFrames", "maxQueuedFrameBytes", "maxConcurrentDownloads", "maxConcurrentPrints", "maxStagedBytes", "maxCopies"}) + { + Object value = limits.opt(name); + if(!(value instanceof Number number) || !number.toString().matches("[0-9]+") || number.longValue() < get(name)) + { + throw new PrintingException(409, "Client cannot support configured realtime limit: " + name); + } + } + } + + + /** Validate a cross-field lower bound. */ + + + private void atLeast(String name, long minimum) + { + if(get(name) < minimum) + { + throw new PrintingException(400, "Realtime settings conflict at " + name); + } + } + + + /** Canonical policy defaults and validation ranges. */ + private static Map fields() + { + Map fields = new LinkedHashMap<>(); + fields.put("maxPdfBytes", new long[] {104857600, 1, Integer.MAX_VALUE}); + fields.put("maxCopies", new long[] {100, 1, 32767}); + fields.put("maxZplBytes", new long[] {20971520, 1, Integer.MAX_VALUE}); + fields.put("maxFrameBytes", new long[] {1048576, 8192, 16777216}); + fields.put("maxQueuedFrames", new long[] {64, 8, 4096}); + fields.put("maxQueuedFrameBytes", new long[] {4194304, 8192, 67108864}); + fields.put("maxConcurrentDownloads", new long[] {1, 1, 16}); + fields.put("maxConcurrentPrints", new long[] {1, 1, 16}); + fields.put("maxStagedBytes", new long[] {536870912, 1, 9007199254740991L}); + fields.put("maxDownloadBytesPerSecond", new long[] {0, 0, Integer.MAX_VALUE}); + fields.put("requestTimeoutMillis", new long[] {30000, 1000, 300000}); + fields.put("downloadConnectTimeoutMillis", new long[] {30000, 1000, 300000}); + fields.put("webSocketHandshakeTimeoutMillis", new long[] {30000, 1000, 300000}); + fields.put("ackTimeoutMillis", new long[] {15000, 1000, 300000}); + fields.put("downloadIdleTimeoutMillis", new long[] {30000, 1000, 3600000}); + fields.put("downloadTimeoutMillis", new long[] {600000, 1000, 86400000}); + fields.put("nativeSubmissionTimeoutMillis", new long[] {120000, 1000, 86400000}); + fields.put("heartbeatIntervalSeconds", new long[] {20, 5, 300}); + fields.put("heartbeatTimeoutSeconds", new long[] {60, 10, 900}); + fields.put("reconnectMinDelayMillis", new long[] {1000, 100, 60000}); + fields.put("reconnectMaxDelayMillis", new long[] {30000, 1000, 300000}); + fields.put("reconnectJitterPercent", new long[] {20, 0, 50}); + return (java.util.Collections.unmodifiableMap(fields)); + } +} diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeProcesses.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeProcesses.java new file mode 100644 index 0000000..064db94 --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeProcesses.java @@ -0,0 +1,256 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.math.BigDecimal; +import java.sql.SQLException; +import java.util.List; +import com.kingsrook.qqq.backend.core.actions.processes.BackendStep; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType; +import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QComponentType; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendComponentMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QFrontendStepMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData; +import org.json.JSONObject; + + +/******************************************************************************* + ** Native tenant settings use human units without exposing connection secrets. + *******************************************************************************/ +public class PrintingRealtimeProcesses implements BackendStep +{ + public static final String SETTINGS = "printingRealtimeSettings"; + public static final String ADMISSION = "printingAdmissionSettings"; + private static final long MEBIBYTE = 1024L * 1024; + private static final List DOCUMENTS = List.of( + new Setting("maxPdfMiB", "maxPdfBytes", "Maximum PDF size (MiB)", MEBIBYTE), + new Setting("maxZplMiB", "maxZplBytes", "Maximum ZPL size (MiB)", MEBIBYTE), + new Setting("maxCopies", "maxCopies", "Maximum copies per job", 1), + new Setting("maxStagedMiB", "maxStagedBytes", "Maximum staged documents on each device (MiB)", MEBIBYTE)); + private static final List TRANSFERS = List.of( + new Setting("maxConcurrentDownloads", "maxConcurrentDownloads", "Concurrent document downloads", 1), + new Setting("maxConcurrentPrints", "maxConcurrentPrints", "Concurrent printer submissions", 1), + new Setting("downloadKiBPerSecond", "maxDownloadBytesPerSecond", "Download bandwidth (KiB/second; 0 means unlimited)", 1024), + new Setting("requestTimeoutSeconds", "requestTimeoutMillis", "Request timeout (seconds)", 1000), + new Setting("downloadConnectSeconds", "downloadConnectTimeoutMillis", "Download connection timeout (seconds)", 1000), + new Setting("downloadIdleSeconds", "downloadIdleTimeoutMillis", "Download idle timeout (seconds)", 1000), + new Setting("downloadTotalSeconds", "downloadTimeoutMillis", "Total download timeout (seconds)", 1000), + new Setting("nativeSubmissionSeconds", "nativeSubmissionTimeoutMillis", "Native printer submission timeout (seconds)", 1000)); + private static final List CONNECTION = List.of( + new Setting("handshakeSeconds", "webSocketHandshakeTimeoutMillis", "WebSocket connection timeout (seconds)", 1000), + new Setting("heartbeatIntervalSeconds", "heartbeatIntervalSeconds", "Connection heartbeat interval (seconds)", 1), + new Setting("heartbeatTimeoutSeconds", "heartbeatTimeoutSeconds", "Unresponsive connection timeout (seconds)", 1), + new Setting("reconnectMinSeconds", "reconnectMinDelayMillis", "Initial reconnect delay (seconds)", 1000), + new Setting("reconnectMaxSeconds", "reconnectMaxDelayMillis", "Maximum reconnect delay (seconds)", 1000), + new Setting("reconnectJitterPercent", "reconnectJitterPercent", "Reconnect jitter (%)", 1), + new Setting("ackTimeoutSeconds", "ackTimeoutMillis", "Acknowledgment timeout (seconds)", 1000)); + private static final List CONTROL = List.of( + new Setting("maxFrameKiB", "maxFrameBytes", "Maximum control message (KiB)", 1024), + new Setting("maxQueuedFrames", "maxQueuedFrames", "Maximum queued control messages", 1), + new Setting("maxQueuedControlMiB", "maxQueuedFrameBytes", "Maximum queued control message data (MiB)", MEBIBYTE)); + private static final List ADMISSION_COUNTS = List.of( + new Setting("deviceOutstandingJobs", "deviceOutstandingJobs", "Outstanding jobs per device", 1), + new Setting("tenantOutstandingJobs", "tenantOutstandingJobs", "Outstanding jobs across this tenant", 1), + new Setting("deviceJobsPerHour", "deviceJobsPerHour", "Jobs allocated per device per hour", 1), + new Setting("tenantJobsPerHour", "tenantJobsPerHour", "Jobs allocated across this tenant per hour", 1), + new Setting("devicePrinters", "devicePrinters", "Printers per device", 1), + new Setting("tenantPrinters", "tenantPrinters", "Printers across this tenant", 1)); + private static final List ADMISSION_STORAGE = List.of( + new Setting("deviceOutstandingMiB", "deviceOutstandingBytes", "Outstanding document data per device (MiB)", MEBIBYTE), + new Setting("tenantOutstandingMiB", "tenantOutstandingBytes", "Outstanding document data across this tenant (MiB)", MEBIBYTE), + new Setting("deviceRetainedGiB", "deviceRetainedBytes", "Retained document data per device (GiB)", 1024 * MEBIBYTE), + new Setting("tenantRetainedGiB", "tenantRetainedBytes", "Retained document data across this tenant (GiB)", 1024 * MEBIBYTE)); + + + + /*************************************************************************** + ** Typed native forms group limits by purpose and reuse normal QQQ permissions. + ***************************************************************************/ + public static void register(QInstance instance, QAppMetaData app) + { + QProcessMetaData process = new QProcessMetaData().withName(SETTINGS).withLabel("Printing connection and limits") + .withPermissionRules(PrintingProcesses.permissionRules()) + .withStep(backend("prepare")) + .withStep(new QFrontendStepMetaData().withName("review").withLabel("Current connection policy") + .withViewFields(List.of(text("transport", "Connection"), text("scope", "Settings scope"), text("revisionLabel", "Current policy"))) + .withComponent(new QFrontendComponentMetaData().withType(QComponentType.VIEW_FORM))) + .withStep(form("documents", "Document limits", DOCUMENTS)) + .withStep(form("transfers", "Document transfers", TRANSFERS)) + .withStep(form("connection", "Connection recovery", CONNECTION)) + .withStep(form("control", "Control channel capacity", CONTROL)) + .withStep(backend("execute")) + .withStep(new QFrontendStepMetaData().withName("result").withLabel("Result").withViewFields(List.of(text("result", "Result"))) + .withComponent(new QFrontendComponentMetaData().withType(QComponentType.VIEW_FORM))); + instance.addProcess(process); + app.withChild(process); + QProcessMetaData admission = new QProcessMetaData().withName(ADMISSION).withLabel("Printing queue and storage limits") + .withPermissionRules(PrintingProcesses.permissionRules()).withStep(backend("prepare")) + .withStep(new QFrontendStepMetaData().withName("review").withLabel("Current admission policy") + .withViewFields(List.of(text("scope", "Settings scope"))).withComponent(new QFrontendComponentMetaData().withType(QComponentType.VIEW_FORM))) + .withStep(form("counts", "Queue, rate and printer limits", ADMISSION_COUNTS)) + .withStep(form("capacity", "Document storage capacity", ADMISSION_STORAGE)) + .withStep(backend("execute")) + .withStep(new QFrontendStepMetaData().withName("result").withLabel("Result").withViewFields(List.of(text("result", "Result"))) + .withComponent(new QFrontendComponentMetaData().withType(QComponentType.VIEW_FORM))); + instance.addProcess(admission); + app.withChild(admission); + } + + + + /*************************************************************************** + ** Each backend entrypoint independently checks the current user and tenant. + ***************************************************************************/ + @Override + public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException + { + PrintingControlAuthorization.principal(); + boolean admission = ADMISSION.equals(input.getProcessName()); + if(!SETTINGS.equals(input.getProcessName()) && !admission) + { + throw new QException("Unknown printing connection settings action"); + } + PrintingService service = PrintingProcesses.currentService(); + try + { + if("prepare".equals(input.getStepName())) + { + JSONObject policy = admission ? service.getAdmissionSettings() : service.getRealtimeSettings(); + for(List group : groups(admission)) + { + for(Setting setting : group) + { + output.addValue(setting.input(), BigDecimal.valueOf(policy.getLong(setting.policy())).divide(BigDecimal.valueOf(setting.unit()))); + } + } + output.addValue("transport", "The device opens a two-way WebSocket to this server. Documents use separate HTTPS transfers; job and command polling is not used."); + output.addValue("scope", admission ? "These limits apply to this tenant. Capacity is checked when jobs and printers are admitted. Changing a limit does not remove existing jobs, documents or history." + : "These settings apply to this tenant. Saved changes are sent to connected devices. Unsupported device limits require an explicit configuration correction; they are not silently reduced."); + output.addValue("revisionLabel", admission ? "Tenant admission policy" : "Policy revision " + policy.getLong("revision")); + } + else if("execute".equals(input.getStepName())) + { + JSONObject requested = new JSONObject(); + for(List group : groups(admission)) + { + for(Setting setting : group) + { + requested.put(setting.policy(), number(input, setting)); + } + } + JSONObject saved = admission ? service.saveAdmissionSettings(requested) : service.saveRealtimeSettings(requested); + output.addValue("result", admission ? "Queue, rate, printer and storage limits saved for this tenant. Existing jobs and history were preserved." + : "Connection and document limits saved as policy revision " + saved.getLong("revision") + ". Connected devices receive the update over their existing WebSocket; no documents or job history were removed."); + } + else + { + throw new QException("Unknown printing connection settings step"); + } + } + catch(SQLException | PrintingException | IllegalArgumentException exception) + { + throw new QException("Connection settings were not saved: " + exception.getMessage(), exception); + } + } + + + + /*************************************************************************** + ** Numeric conversions reject missing, rounded and overflowing input values. + ***************************************************************************/ + private static long number(RunBackendStepInput input, Setting setting) throws QException + { + Object value = input.getValue(setting.input()); + if(value == null) + { + throw new QException("Specify " + setting.label().toLowerCase(java.util.Locale.ROOT)); + } + try + { + return (new BigDecimal(value.toString()).multiply(BigDecimal.valueOf(setting.unit())).longValueExact()); + } + catch(NumberFormatException | ArithmeticException exception) + { + throw new QException(setting.label() + " must be a valid number within the supported range", exception); + } + } + + + + /*************************************************************************** + ** Keep reading and writing the same complete set of advertised policy fields. + ***************************************************************************/ + private static List> groups(boolean admission) + { + return (admission ? List.of(ADMISSION_COUNTS, ADMISSION_STORAGE) : List.of(DOCUMENTS, TRANSFERS, CONNECTION, CONTROL)); + } + + + + /*************************************************************************** + ** Human-readable units remain local to the form, never part of wire identity. + ***************************************************************************/ + private static QFrontendStepMetaData form(String name, String label, List settings) + { + return (new QFrontendStepMetaData().withName(name).withLabel(label) + .withFormFields(settings.stream().map(setting -> new QFieldMetaData(setting.input(), setting.unit() == 1 ? QFieldType.INTEGER : QFieldType.DECIMAL) + .withLabel(setting.label()).withIsRequired(true)).toList()) + .withComponent(new QFrontendComponentMetaData().withType(QComponentType.EDIT_FORM))); + } + + + + /*************************************************************************** + ** Every step uses the same native backend and read-only result widgets. + ***************************************************************************/ + private static QBackendStepMetaData backend(String name) + { + return (new QBackendStepMetaData().withName(name).withCode(new QCodeReference(PrintingRealtimeProcesses.class))); + } + + + + /*************************************************************************** + ** View fields contain policy descriptions rather than editable identifiers. + ***************************************************************************/ + private static QFieldMetaData text(String name, String label) + { + return (new QFieldMetaData(name, QFieldType.STRING).withLabel(label)); + } + + + + /*************************************************************************** + ** Unit factors make the form reversible without losing byte precision. + ***************************************************************************/ + private record Setting(String input, String policy, String label, long unit) + { + } +} diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java new file mode 100644 index 0000000..c70393a --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRealtimeRoutes.java @@ -0,0 +1,669 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.ArrayDeque; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import com.kingsrook.qqq.backend.core.actions.profile.UserProfiles; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException; +import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.permissions.ClientChannelsMetaData; +import com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModuleDispatcher; +import com.kingsrook.qqq.middleware.javalin.QJavalinRouteProviderInterface; +import com.kingsrook.qqq.middleware.javalin.executors.ExecutorSessionUtils; +import io.javalin.config.JavalinConfig; +import io.javalin.http.Context; +import io.javalin.http.HttpResponseException; +import io.javalin.websocket.WsContext; +import org.eclipse.jetty.websocket.api.Callback; +import org.json.JSONArray; +import org.json.JSONObject; + + +/** Current-user authenticated outbound channels. SQL is authoritative; notifications are only wakeups. */ +public final class PrintingRealtimeRoutes implements QJavalinRouteProviderInterface, AutoCloseable +{ + public static final String PATH = "/qqq/v1/qbit/printing/device/realtime"; + public static final String SUBPROTOCOL = "qqq-printing.v1"; + private final PrintingService service; + private final Map channels = new ConcurrentHashMap<>(); + private final ExecutorService workers = Executors.newVirtualThreadPerTaskExecutor(); + private final AutoCloseable subscription; + private QInstance instance; + + + /** Register this provider alongside the host's ordinary native QQQ endpoints. */ + public PrintingRealtimeRoutes(PrintingService service) + { + this.service = java.util.Objects.requireNonNull(service); + service.withRealtimeClaimGuard(this::requireReady); + subscription = service.getRealtimeBus().subscribe((tenant, device) -> + { + if(service.getTenantId().equals(tenant)) + { + channels.values().stream().filter(channel -> device == null || device.equals(channel.device)).forEach(Channel::changed); + } + }); + } + + + @Override + public void setQInstance(QInstance value) + { + instance = value; + } + + + @Override + public void acceptJavalinConfig(JavalinConfig config) + { + config.routes.wsBeforeUpgrade(PATH, context -> + { + try + { + validateUpgrade(context); + authorize(context); + String cookieName = context.cookie("sessionId") != null && !context.cookie("sessionId").isBlank() ? "sessionId" : "sessionUUID"; + context.attribute("printingRealtimeCredential", new Credential(cookieName, context.cookie(cookieName), context.header(PrintingAgentAuthorization.TOKEN_HEADER), context.header("X-QQQ-Client"))); + } + catch(Exception exception) + { + throw new HttpResponseException(exception instanceof QAuthenticationException ? 401 : 403, "Realtime authentication or origin denied", Map.of()); + } + finally + { + QContext.clear(); + } + }); + config.routes.ws(PATH, socket -> + { + socket.onConnect(context -> + { + String device = context.header(PrintingAgentAuthorization.DEVICE_HEADER); + Channel next = new Channel(context, device, context.attribute("printingRealtimeCredential")); + next.policy = service.realtimePolicy(); + Channel previous = channels.put(device, next); + if(previous != null) + { + previous.stop("Connection replaced"); + } + service.getRealtimeBus().acquire(service.getTenantId(), device, next.generation, () -> next.stop("Connection replaced")); + context.session.setMaxTextMessageSize(next.policy.get("maxFrameBytes")); + context.session.setMaxBinaryMessageSize(0); + context.session.setIdleTimeout(Duration.ofMillis(next.policy.get("webSocketHandshakeTimeoutMillis"))); + }); + socket.onMessage(context -> + { + Channel channel = current(context); + if(channel != null) + { + channel.accept(context.message()); + } + }); + socket.onBinaryMessage(context -> context.closeSession(1003, "Only JSON text is supported")); + socket.onClose(context -> remove(context)); + socket.onError(context -> remove(context)); + }); + config.events.serverStopping(this::close); + } + + + /** Resolve only the current socket generation. */ + + + private Channel current(WsContext context) + { + Channel channel = channels.get(context.header(PrintingAgentAuthorization.DEVICE_HEADER)); + return (channel != null && channel.socket.session == context.session ? channel : null); + } + + + /** Remove only the closing socket generation. */ + + + private void remove(WsContext context) + { + Channel channel = current(context); + if(channel != null) + { + channel.stop("Connection closed"); + } + } + + + /** Gate native claims on acknowledged current settings. */ + + + private void requireReady(String device) throws java.sql.SQLException + { + Channel channel = channels.get(device); + if(channel == null || !channel.ready(service.getRealtimeSettings().getLong("revision"))) + { + throw new PrintingException(409, "An authenticated realtime connection must apply current state and settings before claiming work"); + } + } + + + /** Reject query credentials, non-session authentication and foreign origins. */ + + + private static void validateUpgrade(Context context) + { + if(context.queryString() != null || !SUBPROTOCOL.equals(context.header("Sec-WebSocket-Protocol")) + || ((context.cookie("sessionUUID") == null || context.cookie("sessionUUID").isBlank()) && (context.cookie("sessionId") == null || context.cookie("sessionId").isBlank())) + || context.header("Authorization") != null || context.header("x-api-key") != null || context.header("X-QQQ-In-Process-Session") != null) + { + throw new PrintingException(403, "A standard QQQ session and the printing subprotocol are required"); + } + URI request = URI.create(context.url()); + URI origin = URI.create(java.util.Objects.toString(context.header("Origin"), "")); + boolean loopback = Set.of("localhost", "127.0.0.1", "[::1]", "::1").contains(request.getHost()); + if(!("https".equals(request.getScheme()) || (loopback && "http".equals(request.getScheme()))) + || !java.util.Objects.equals(request.getScheme(), origin.getScheme()) + || !java.util.Objects.equals(request.getRawAuthority(), origin.getRawAuthority()) + || origin.getRawUserInfo() != null || origin.getRawQuery() != null || origin.getRawFragment() != null + || !(origin.getPath().isEmpty() || "/".equals(origin.getPath()))) + { + throw new PrintingException(403, "Realtime origin must match the authenticated server origin"); + } + } + + + /** Revalidate the standard current QQQ session and secondary device ownership. */ + + + private void authorize(Context context) throws Exception + { + ExecutorSessionUtils.setupSession(context, instance); + PrintingAgentAuthorization.device(context.header(PrintingAgentAuthorization.DEVICE_HEADER), context.header(PrintingAgentAuthorization.TOKEN_HEADER), false); + } + + + /** Only transport credentials survive upgrade; user roles and session authority are never cached. */ + private static final class Credential + { + private final String cookie; + private final String cookieName; + private final String proof; + private final String clientChannel; + + + /** Capture immutable authenticated transport inputs before Jetty releases the HTTP request. */ + private Credential(String cookieName, String cookie, String proof, String clientChannel) + { + this.cookieName = cookieName; + this.cookie = cookie; + this.proof = proof; + this.clientChannel = clientChannel; + } + + + @Override + public String toString() + { + return ("Printing realtime credential [redacted]"); + } + } + + + @Override + public void close() + { + channels.values().forEach(channel -> channel.stop("Server stopping")); + try + { + subscription.close(); + } + catch(Exception exception) + { + throw new IllegalStateException("Unable to remove realtime subscription", exception); + } + finally + { + workers.shutdown(); + } + } + + + /** One serial bounded mailbox per socket; network waits never run on the committing caller. */ + private final class Channel + { + private final WsContext socket; + private final String device; + private final String generation = UUID.randomUUID().toString(); + private final Credential credential; + private final ArrayDeque mailbox = new ArrayDeque<>(); + private PrintingRealtimePolicy policy = PrintingRealtimePolicy.defaults(); + private JSONObject hello; + private boolean running; + private boolean dirty; + private boolean stopped; + private long queuedBytes; + private long appliedState = -1; + private long appliedSettings = -1; + private long sentState = -1; + private long sentSettings = -1; + + + /** Bind one mailbox to its authenticated socket. */ + + + private Channel(WsContext socket, String device, Credential credential) + { + this.socket = socket; + this.device = device; + this.credential = credential; + } + + + /** Re-enter the configured native authentication module for every operation after HTTP upgrade. */ + private void authorizeCurrent() throws Exception + { + QContext.init(instance, null); + var module = new QAuthenticationModuleDispatcher().getQModule(instance.getAuthentication()); + var session = module.createSession(instance, Map.of(credential.cookieName, credential.cookie)); + UserProfiles.applyIfInstalled(instance, session); + QContext.init(instance, session, null, null); + if(session.getCredentialScope() != null) + { + throw new QAuthenticationException("Realtime printing requires a current user session"); + } + String channel = credential.clientChannel == null || credential.clientChannel.isBlank() ? null : credential.clientChannel.strip(); + if(channel != null && (!ClientChannelsMetaData.of(instance).isAllowedChannel(ClientChannelsMetaData.channelName(channel)) || channel.length() > 200)) + { + throw new QPermissionDeniedException("The current client channel is not allowed"); + } + session.setClientChannel(channel == null ? null : channel.strip()); + PrintingAgentAuthorization.device(device, credential.proof, false); + } + + + /** Read synchronized connection application state. */ + + + private synchronized boolean ready(long revision) + { + return (!stopped && socket.session.isOpen() && service.getRealtimeBus().owns(service.getTenantId(), device, generation) + && hello != null && appliedState == sentState && appliedSettings == sentSettings && appliedSettings == revision); + } + + + /** Bound inbound work before scheduling any database operation. */ + + + private synchronized void accept(String frame) + { + int bytes = frame.getBytes(StandardCharsets.UTF_8).length; + if(bytes > policy.get("maxFrameBytes") || mailbox.size() >= policy.get("maxQueuedFrames") || queuedBytes + bytes > policy.get("maxQueuedFrameBytes")) + { + stop("Realtime mailbox limit exceeded"); + return; + } + mailbox.add(frame); + queuedBytes += bytes; + schedule(); + } + + + /** Coalesce commit wakeups without blocking the committing caller. */ + + + private synchronized void changed() + { + dirty = true; + schedule(); + } + + + /** Start at most one serial worker for this connection. */ + + + private void schedule() + { + if(!running && !stopped) + { + running = true; + workers.execute(this::drain); + } + } + + + /** Process bounded inbound work and coalesced durable-state wakeups. */ + + + private void drain() + { + while(true) + { + String frame; + synchronized(this) + { + frame = mailbox.poll(); + if(frame != null) + { + queuedBytes -= frame.getBytes(StandardCharsets.UTF_8).length; + } + else if(!dirty || hello == null || stopped) + { + running = false; + return; + } + else + { + dirty = false; + } + } + try + { + if(channels.get(device) != this || stopped || !service.getRealtimeBus().owns(service.getTenantId(), device, generation)) + { + return; + } + authorizeCurrent(); + if(frame == null) + { + snapshot(false); + } + else + { + receive(new JSONObject(frame)); + } + } + catch(Exception exception) + { + if(reject(frame, exception)) + { + return; + } + } + finally + { + QContext.clear(); + } + } + } + + + /** Apply one authenticated protocol operation. */ + + + private void receive(JSONObject frame) throws Exception + { + if(!frame.keySet().equals(Set.of("type", "messageId", "payload")) || !(frame.opt("messageId") instanceof String messageId) + || !messageId.matches("[A-Za-z0-9._-]{1,200}") || !(frame.opt("payload") instanceof JSONObject payload)) + { + throw new PrintingException(400, "Invalid realtime envelope"); + } + String type = frame.getString("type"); + if(hello == null && !"HELLO".equals(type)) + { + throw new PrintingException(409, "HELLO is required first"); + } + switch(type) + { + case "HELLO" -> + { + if(hello != null || !payload.keySet().equals(Set.of("agentVersion", "capabilities", "lastAppliedRevision", "lastSettingsRevision", "limits")) + || !payload.getJSONArray("capabilities").toList().contains("realtime-v1") + || payload.getJSONArray("capabilities").length() > 3 + || !Set.of("realtime-v1", "REFRESH_INVENTORY", "CANCEL_SPOOL").containsAll(payload.getJSONArray("capabilities").toList())) + { + throw new PrintingException(409, "Realtime capability is required"); + } + hello = payload; + snapshot(true); + return; + } + case "STATE_APPLIED" -> + { + long state = integer(payload, "lastAppliedRevision"); + long settings = integer(payload, "lastSettingsRevision"); + if(state > sentState || settings > sentSettings || state < appliedState || settings < appliedSettings) + { + throw new PrintingException(409, "Only the current delivered state and settings may be acknowledged"); + } + service.applyRealtimeState(device, controlRequest(state), settings); + synchronized(this) + { + appliedState = state; + appliedSettings = settings; + } + changed(); + } + case "JOB_EVENT" -> + { + String job = payload.getString("jobId"); + PrintingControlAuthorization.job(device, job); + service.event(device, job, payload.getJSONObject("event")); + } + case "COMMAND_RECEIPT" -> + { + String command = payload.getString("commandId"); + PrintingControlAuthorization.table(PrintingTables.COMMAND); + service.getDeviceCommand(device, command); + PrintingUiApiSpec.command(device, command); + service.deviceControlReceipt(device, command, payload.getJSONObject("receipt")); + changed(); + } + case "HEARTBEAT" -> service.heartbeat(device, payload); + case "INVENTORY" -> service.inventory(device, payload.getJSONArray("printers"), PrintingAgentAuthorization.inventory(device)); + case "PING" -> + { + send("PONG", new JSONObject().put("serverTime", Instant.now().toString())); + return; + } + default -> throw new PrintingException(400, "Unsupported realtime message"); + } + send("ACK", new JSONObject().put("acknowledgedMessageId", messageId)); + } + + + /** Domain errors retain their identity and never masquerade as a durable acknowledgment. */ + private boolean reject(String request, Exception exception) + { + System.getLogger(PrintingRealtimeRoutes.class.getName()).log(System.Logger.Level.WARNING, + "Realtime operation rejected: " + exception.getClass().getSimpleName() + " at " + (exception.getStackTrace().length == 0 ? "unknown" : exception.getStackTrace()[0])); + boolean authorization = exception instanceof QAuthenticationException || exception instanceof QPermissionDeniedException; + int status = exception instanceof PrintingException printing ? printing.getStatus() : authorization ? 403 : exception instanceof org.json.JSONException ? 400 : 500; + boolean connectionError = request == null || hello == null || sentSettings < 0 || status == 401 || status == 403 || status >= 500; + String code = switch(status) + { + case 401, 403 -> "AUTHORIZATION_REQUIRED"; + case 404 -> "NOT_FOUND"; + case 410 -> "EXPIRED"; + case 409 -> connectionError ? "CONFIGURATION_REQUIRED" : "CONFLICT"; + case 400, 413 -> "INVALID_MESSAGE"; + default -> "OPERATION_FAILED"; + }; + String id = UUID.randomUUID().toString(); + try + { + if(request != null) + { + String requested = new JSONObject(request).optString("messageId"); + if(requested.matches("[A-Za-z0-9._-]{1,200}")) + { + id = requested; + } + } + String error = new JSONObject().put("type", "ERROR").put("messageId", id).put("payload", new JSONObject().put("code", code) + .put("message", "The realtime operation was rejected; inspect the current state before retrying").put("retryable", status >= 500)).toString(); + CompletableFuture sent = new CompletableFuture<>(); + socket.session.sendText(error, Callback.from(() -> sent.complete(null), sent::completeExceptionally)); + sent.get(policy.get("ackTimeoutMillis"), TimeUnit.MILLISECONDS); + } + catch(Exception failure) + { + connectionError = true; + } + if(connectionError) + { + stop("Realtime operation rejected"); + } + return (connectionError); + } + + + /** Reuse the durable control protocol with advertised supported operations. */ + + + private JSONObject controlRequest(long state) + { + JSONArray capabilities = new JSONArray(); + for(Object value : hello.getJSONArray("capabilities")) + { + if(Set.of("REFRESH_INVENTORY", "CANCEL_SPOOL").contains(value)) + { + capabilities.put(value); + } + } + return (new JSONObject().put("controlVersion", 1).put("agentVersion", hello.getString("agentVersion")) + .put("capabilities", capabilities).put("lastAppliedRevision", state)); + } + + + /** Reconstruct current state and pending commands after registration or commit. */ + + + private void snapshot(boolean initial) throws Exception + { + JSONObject settings = service.getRealtimeSettings(); + if(initial && integer(hello, "lastSettingsRevision") > settings.getLong("revision")) + { + throw new PrintingException(409, "Unknown realtime settings revision"); + } + JSONObject values = new JSONObject(settings.toString()); + values.remove("revision"); + policy = PrintingRealtimePolicy.fromJson(values); + policy.requireClientLimits(hello.getJSONObject("limits")); + socket.session.setMaxTextMessageSize(policy.get("maxFrameBytes")); + socket.session.setIdleTimeout(Duration.ofSeconds(policy.get("heartbeatTimeoutSeconds"))); + JSONObject state = service.deviceControlPoll(device, controlRequest(initial ? integer(hello, "lastAppliedRevision") : Math.max(0, appliedState))); + PrintingControlAuthorization.table(PrintingTables.COMMAND); + JSONObject desired = state.getJSONObject("desiredState"); + long previousState; + long previousSettings; + synchronized(this) + { + previousState = sentState; + previousSettings = sentSettings; + sentState = desired.getLong("revision"); + sentSettings = settings.getLong("revision"); + } + if(initial) + { + send("WELCOME", new JSONObject().put("connectionId", UUID.randomUUID().toString()).put("serverTime", Instant.now().toString()) + .put("desiredState", desired).put("settings", settings)); + } + else + { + if(previousSettings != sentSettings) + { + send("SETTINGS", settings); + } + if(previousState != sentState) + { + send("DESIRED_STATE", desired); + } + } + if(!ready(sentSettings)) + { + return; + } + for(Object item : state.getJSONArray("commands")) + { + JSONObject command = (JSONObject) item; + PrintingUiApiSpec.command(device, command.getString("commandId")); + send("COMMAND", command); + } + send("JOB_AVAILABLE", new JSONObject()); + } + + + /** Revalidate authority and await one bounded asynchronous socket write. */ + + + private void send(String type, JSONObject payload) throws Exception + { + authorizeCurrent(); + synchronized(this) + { + if(stopped || channels.get(device) != this || !service.getRealtimeBus().owns(service.getTenantId(), device, generation)) + { + throw new PrintingException(409, "Connection was replaced"); + } + } + String frame = new JSONObject().put("type", type).put("messageId", UUID.randomUUID().toString()).put("payload", payload).toString(); + if(frame.getBytes(StandardCharsets.UTF_8).length > policy.get("maxFrameBytes")) + { + throw new PrintingException(413, "Realtime frame limit exceeded"); + } + CompletableFuture sent = new CompletableFuture<>(); + socket.session.sendText(frame, Callback.from(() -> sent.complete(null), sent::completeExceptionally)); + sent.get(policy.get("ackTimeoutMillis"), TimeUnit.MILLISECONDS); + } + + + /** Fence and remove this connection without affecting a replacement. */ + + + private synchronized void stop(String reason) + { + if(!stopped) + { + stopped = true; + mailbox.clear(); + queuedBytes = 0; + channels.remove(device, this); + service.getRealtimeBus().release(service.getTenantId(), device, generation); + socket.closeSession(1008, reason); + } + } + } + + + /** Reject absent, fractional and negative revision values. */ + + + private static long integer(JSONObject value, String name) + { + Object raw = value.get(name); + if(!(raw instanceof Number number) || !number.toString().matches("[0-9]+")) + { + throw new PrintingException(400, "Expected nonnegative integer " + name); + } + return (number.longValue()); + } +} diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRetentionProcesses.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRetentionProcesses.java index 4f0cc08..785af57 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingRetentionProcesses.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRetentionProcesses.java @@ -63,6 +63,7 @@ public class PrintingRetentionProcesses implements BackendStep .withStep(review("Current retention and storage")) .withStep(new QFrontendStepMetaData().withName("settings").withLabel("Edit retention") .withFormFields(List.of(new QFieldMetaData("days", QFieldType.INTEGER).withLabel("Keep successful job documents for days (1–36500)").withIsRequired(true), + new QFieldMetaData("batchSize", QFieldType.INTEGER).withLabel("Maximum documents per cleanup run (1–200)").withIsRequired(true), new QFieldMetaData("enabled", QFieldType.BOOLEAN).withLabel("Enable document retention cleanup").withIsRequired(true))) .withComponent(new QFrontendComponentMetaData().withType(QComponentType.EDIT_FORM))) .withStep(backend("execute")).withStep(result()); @@ -165,11 +166,12 @@ public class PrintingRetentionProcesses implements BackendStep { Integer days = input.getValueInteger("days"); Boolean enabled = input.getValueBoolean("enabled"); - if(days == null || enabled == null) + Integer batchSize = input.getValueInteger("batchSize"); + if(days == null || enabled == null || batchSize == null) { throw new QException("Specify retention days and whether cleanup is enabled"); } - service.saveRetentionSettings(days, enabled); + service.saveRetentionSettings(days, enabled, batchSize); output.addValue("result", "Printing retention settings saved for this tenant. Changing the policy does not run cleanup immediately."); } else if("execute".equals(input.getStepName()) && (RUN_NOW.equals(input.getProcessName()) || scheduled)) @@ -198,6 +200,7 @@ public class PrintingRetentionProcesses implements BackendStep JSONObject settings = service.getRetentionSettings(); output.addValue("days", settings.getInt("days")); output.addValue("enabled", settings.getBoolean("enabled")); + output.addValue("batchSize", settings.getInt("batchSize")); output.addValue("policy", (settings.getBoolean("enabled") ? "Enabled" : "Disabled") + ": retain successful job documents for " + settings.getInt("days") + " days; at most " + settings.getInt("batchSize") + " documents per run."); String table = settings.optString("storageTableName", ""); String label = QContext.getQInstance().getTable(table) == null ? table : QContext.getQInstance().getTable(table).getLabel(); diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingRoutes.java b/src/main/java/com/kingsrook/qbits/printing/PrintingRoutes.java index b17ae44..485390b 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingRoutes.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingRoutes.java @@ -224,9 +224,12 @@ public class PrintingRoutes implements QJavalinRouteProviderInterface /*************************************************************************** ** Parse bounded protocol JSON; the server also configures request limits. ***************************************************************************/ - private static JSONObject body(Context context) + private JSONObject body(Context context) throws java.sql.SQLException { - if(context.bodyAsBytes().length > 29 * 1024 * 1024) + PrintingRealtimePolicy policy = service.realtimePolicy(); + long maximum = context.path().equals(BASE + "/admin/jobs") + ? ((Math.max(policy.get("maxPdfBytes"), policy.get("maxZplBytes")) + 2) / 3) * 4 + 65536 : 2097152; + if((context.req().getContentLengthLong() > maximum) || context.bodyAsBytes().length > maximum) { throw new PrintingException(413, "Request is too large"); } diff --git a/src/main/java/com/kingsrook/qbits/printing/PrintingService.java b/src/main/java/com/kingsrook/qbits/printing/PrintingService.java index 1b89b26..e82338c 100644 --- a/src/main/java/com/kingsrook/qbits/printing/PrintingService.java +++ b/src/main/java/com/kingsrook/qbits/printing/PrintingService.java @@ -47,7 +47,6 @@ import org.json.JSONObject; public class PrintingService { static final String CANCELED_BEFORE_DISPATCH = "Canceled by operator before dispatch"; - private static final int MAX_CONTENT_BYTES = 20 * 1024 * 1024; private static final String JOB_COLUMNS = "id, tenant_id, device_id, printer_id, title, format, content_hash, content_length, options_json, idempotency_key, created_at, status, message, spooler_job_id, printer_row_id"; private final DataSource dataSource; private final String tenantId; @@ -55,6 +54,121 @@ public class PrintingService private PrintingAdmissionLimits admissionLimits = PrintingAdmissionLimits.defaults(); private String storageTableName; private PrintingRetentionPolicy defaultRetentionPolicy = new PrintingRetentionPolicy(true, 365, 200); + private PrintingRealtimeBus realtimeBus = new PrintingRealtimeBus(); + private RealtimeClaimGuard realtimeClaimGuard = device -> + { + throw new PrintingException(503, "The host has not registered the realtime printing route provider"); + }; + + + /** Native claims require a synchronized current-user channel, not a polling fallback. */ + @FunctionalInterface + public interface RealtimeClaimGuard + { + /** Check this connection's authoritative applied state. */ + void requireReady(String device) throws SQLException; + } + + + /** Route-provider lifecycle owns the readiness guard. */ + void withRealtimeClaimGuard(RealtimeClaimGuard guard) + { + realtimeClaimGuard = guard; + } + + + /** Enforce readiness on native HTTP claims before selecting any job. */ + public void requireRealtimeReady(String device) throws SQLException + { + realtimeClaimGuard.requireReady(device); + } + + + /** Persist application observations before the WebSocket acknowledgment is emitted. */ + void applyRealtimeState(String device, JSONObject control, long settings) throws SQLException + { + if(settings > getRealtimeSettings().getLong("revision")) + { + throw new PrintingException(409, "Realtime settings changed before acknowledgment"); + } + deviceControlPoll(device, control); + try(Connection connection = dataSource.getConnection()) + { + if(update(connection, "UPDATE printing_device SET realtime_applied_revision=CASE WHEN realtime_applied_revision=?)", settings, settings, tenantId, device, tenantId, settings) != 1) + { + throw new PrintingException(409, "Realtime settings changed before acknowledgment"); + } + } + } + + + /** A host may share this event bus across services using the same database. */ + public PrintingService withRealtimeBus(PrintingRealtimeBus bus) + { + realtimeBus = java.util.Objects.requireNonNull(bus); + return (this); + } + + + /** Instance-owned notifications never contain credentials or document bytes. */ + public PrintingRealtimeBus getRealtimeBus() + { + return (realtimeBus); + } + + + /** Complete persisted tenant settings; defaults apply until the first administrator edit. */ + public JSONObject getRealtimeSettings() throws SQLException + { + ensureRetentionSettings(); + try(Connection connection = dataSource.getConnection()) + { + JSONObject row = query(connection, "SELECT realtime_policy_json, realtime_revision FROM printing_settings WHERE tenant_id=?", tenantId).getFirst(); + PrintingRealtimePolicy policy = row.isNull("realtime_policy_json") ? PrintingRealtimePolicy.defaults() + : PrintingRealtimePolicy.fromJson(new JSONObject(row.getString("realtime_policy_json"))); + return (policy.toJson(row.getLong("realtime_revision"))); + } + } + + + /** The native adapter authorizes the administrator before this atomic update. */ + public JSONObject saveRealtimeSettings(JSONObject request) throws SQLException + { + PrintingRealtimePolicy policy = PrintingRealtimePolicy.fromJson(request); + ensureRetentionSettings(); + JSONObject result; + try(Connection connection = dataSource.getConnection()) + { + connection.setAutoCommit(false); + try + { + long revision = query(connection, "SELECT realtime_revision FROM printing_settings WHERE tenant_id=? FOR UPDATE", tenantId).getFirst().getLong("realtime_revision") + 1; + if(revision < 1 || revision > 9007199254740991L) + { + throw new PrintingException(409, "Realtime settings revision capacity is exhausted"); + } + update(connection, "UPDATE printing_settings SET realtime_policy_json=?, realtime_revision=? WHERE tenant_id=?", request.toString(), revision, tenantId); + connection.commit(); + result = policy.toJson(revision); + } + catch(Exception exception) + { + connection.rollback(); + throw exception; + } + } + realtimeBus.publish(tenantId, null); + return (result); + } + + + /** Resolve immutable validated policy values for one request. */ + public PrintingRealtimePolicy realtimePolicy() throws SQLException + { + JSONObject settings = getRealtimeSettings(); + settings.remove("revision"); + return (PrintingRealtimePolicy.fromJson(settings)); + } @@ -68,6 +182,24 @@ public class PrintingService } + /** Native tenant administrator admission settings. */ + public JSONObject getAdmissionSettings() throws SQLException + { + ensureRetentionSettings(); + return (new PrintingAdmissionSettings(dataSource, tenantId, admissionLimits).get()); + } + + + /** Save validated tenant overrides without changing retained job receipts. */ + public JSONObject saveAdmissionSettings(JSONObject request) throws SQLException + { + ensureRetentionSettings(); + JSONObject result = new PrintingAdmissionSettings(dataSource, tenantId, admissionLimits).save(request); + realtimeBus.publish(tenantId, null); + return (result); + } + + /*************************************************************************** ** Bind a host-owned storage table before exposing this service to requests. @@ -147,6 +279,10 @@ public class PrintingService "CREATE TABLE IF NOT EXISTS printing_settings (tenant_id VARCHAR(200) PRIMARY KEY, retention_days INTEGER NOT NULL, retention_enabled INTEGER NOT NULL, last_run_at VARCHAR(40), last_run_count INTEGER, last_run_error VARCHAR(500))", "ALTER TABLE printing_settings ADD COLUMN IF NOT EXISTS lease_token VARCHAR(36)", "ALTER TABLE printing_settings ADD COLUMN IF NOT EXISTS lease_expires_at VARCHAR(40)", + "ALTER TABLE printing_settings ADD COLUMN IF NOT EXISTS realtime_policy_json VARCHAR(8000)", + "ALTER TABLE printing_settings ADD COLUMN IF NOT EXISTS realtime_revision BIGINT DEFAULT 0 NOT NULL", + "ALTER TABLE printing_settings ADD COLUMN IF NOT EXISTS admission_policy_json VARCHAR(8000)", + "ALTER TABLE printing_settings ADD COLUMN IF NOT EXISTS retention_batch_size INTEGER", "CREATE TABLE IF NOT EXISTS printing_document_orphan (id VARCHAR(36) PRIMARY KEY, tenant_id VARCHAR(200) NOT NULL, storage_table VARCHAR(200) NOT NULL, reference VARCHAR(500) NOT NULL, binding VARCHAR(64) NOT NULL, created_at VARCHAR(40) NOT NULL)", "CREATE INDEX IF NOT EXISTS printing_orphan_tenant_idx ON printing_document_orphan(tenant_id, created_at, id)", "UPDATE printing_job SET printer_row_id=(SELECT p.id FROM printing_printer p WHERE p.device_id=printing_job.device_id AND p.printer_id=printing_job.printer_id AND p.tenant_id=printing_job.tenant_id) WHERE printer_row_id IS NULL", @@ -154,6 +290,7 @@ public class PrintingService "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS display_name VARCHAR(512)", "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS description VARCHAR(4096)", "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS location VARCHAR(512)", + "ALTER TABLE printing_device ADD COLUMN IF NOT EXISTS realtime_applied_revision BIGINT DEFAULT -1 NOT NULL", "ALTER TABLE printing_printer ADD COLUMN IF NOT EXISTS queue_name VARCHAR(1024)", "ALTER TABLE printing_printer ADD COLUMN IF NOT EXISTS display_name VARCHAR(512)", "ALTER TABLE printing_printer ADD COLUMN IF NOT EXISTS description VARCHAR(4096)", @@ -227,14 +364,18 @@ public class PrintingService ***************************************************************************/ public JSONObject deviceAction(String deviceId, JSONObject request, String actor) throws SQLException { - return (new PrintingDeviceControls(dataSource, tenantId).action(deviceId, request, actor)); + JSONObject result = new PrintingDeviceControls(dataSource, tenantId).action(deviceId, request, actor); + realtimeBus.publish(tenantId, deviceId); + return (result); } /** Native authorization supplies the exact reviewed queue identity set, never a predicate executed inside SQL locks. */ JSONObject deviceAction(String deviceId, JSONObject request, String actor, Set authorizedQueue) throws SQLException { - return (new PrintingDeviceControls(dataSource, tenantId).action(deviceId, request, actor, authorizedQueue)); + JSONObject result = new PrintingDeviceControls(dataSource, tenantId).action(deviceId, request, actor, authorizedQueue); + realtimeBus.publish(tenantId, deviceId); + return (result); } @@ -604,7 +745,7 @@ public class PrintingService { incoming.add(required(printers.getJSONObject(index).optString("printerId"), "printerId")); } - PrintingAdmission.inventory(connection, tenantId, deviceId, incoming, admissionLimits); + PrintingAdmission.inventory(connection, tenantId, deviceId, incoming, new PrintingAdmissionSettings(dataSource, tenantId, admissionLimits).effective(connection)); update(connection, "UPDATE printing_printer SET status='UNKNOWN', available=0 WHERE device_id=? AND available=1", deviceId); for(int index = 0; index < printers.length(); index++) { @@ -656,6 +797,7 @@ public class PrintingService throw new PrintingException(404, "Printer not found"); } } + realtimeBus.publish(tenantId, deviceId); } @@ -666,6 +808,11 @@ public class PrintingService public JSONObject submit(JSONObject request) throws SQLException { String format = required(request.optString("format"), "format"); + long maximum = realtimePolicy().documentLimit(format); + if(request.getString("contentBase64").length() > ((maximum + 2) / 3) * 4) + { + throw new PrintingException(413, "Encoded document exceeds its configured format limit"); + } byte[] content; try { @@ -675,13 +822,13 @@ public class PrintingService { throw new PrintingException(400, "Invalid document encoding"); } - if(content.length > MAX_CONTENT_BYTES) + if(content.length > maximum) { - throw new PrintingException(413, "Document exceeds 20971520 bytes"); + throw new PrintingException(413, "Document exceeds configured " + format + " limit of " + maximum + " bytes"); } if(content.length == 0) { - throw new PrintingException(400, "Document must contain 1 to 20971520 bytes"); + throw new PrintingException(400, "Document must contain at least one byte"); } if("PDF".equals(format) && !new String(content, 0, Math.min(5, content.length), StandardCharsets.US_ASCII).equals("%PDF-")) { @@ -721,6 +868,7 @@ public class PrintingService throw new PrintingException(400, "Only PDF and ZPL are supported"); } JSONObject options = printOptions(request); + long contentLimit = realtimePolicy().documentLimit(format); String id = UUID.randomUUID().toString(); String reference = storageTableName == null ? null : documentStorage.prefix() + id + "." + format.toLowerCase(java.util.Locale.ROOT); String binding = storageTableName == null ? null : documentStorage.binding(storageTableName); @@ -774,7 +922,11 @@ public class PrintingService { throw new PrintingException(400, "Printer is not enabled for this format"); } - PrintingAdmission.reserve(connection, tenantId, deviceId, id, contentLength, admissionLimits); + if(contentLength > contentLimit) + { + throw new PrintingException(413, "Document exceeds the current configured format limit"); + } + PrintingAdmission.reserve(connection, tenantId, deviceId, id, contentLength, new PrintingAdmissionSettings(dataSource, tenantId, admissionLimits).effective(connection)); if(storageTableName != null) { reserveUpload(connection, id, storageTableName, reference, binding, deviceId, contentLength, key); @@ -806,6 +958,7 @@ public class PrintingService update(connection, "DELETE FROM printing_document_orphan WHERE id=? AND tenant_id=?", id, tenantId); JSONObject result = job(query(connection, "SELECT " + JOB_COLUMNS + " FROM printing_job WHERE id=?", id).getFirst()); connection.commit(); + realtimeBus.publish(tenantId, deviceId); return (result); } catch(Exception exception) @@ -832,7 +985,7 @@ public class PrintingService /*************************************************************************** ** Preserve request intent: reject malformed types instead of coercing them. ***************************************************************************/ - private static JSONObject printOptions(JSONObject request) + private JSONObject printOptions(JSONObject request) throws SQLException { if(request.has("options") && !(request.opt("options") instanceof JSONObject)) { @@ -845,7 +998,7 @@ public class PrintingService { if(!(copies instanceof Number number)) { - throw new PrintingException(400, "Copies must be an integer between 1 and 100"); + throw new PrintingException(400, "Copies must be an integer within the configured limit"); } try { @@ -853,11 +1006,11 @@ public class PrintingService } catch(ArithmeticException | NumberFormatException exception) { - throw new PrintingException(400, "Copies must be an integer between 1 and 100"); + throw new PrintingException(400, "Copies must be an integer within the configured limit"); } } Object scale = options.has("scale") ? options.opt("scale") : "ACTUAL_SIZE"; - if(count < 1 || count > 100 || !(scale instanceof String) || !Set.of("ACTUAL_SIZE", "FIT").contains(scale)) + if(count < 1 || count > realtimePolicy().get("maxCopies") || !(scale instanceof String) || !Set.of("ACTUAL_SIZE", "FIT").contains(scale)) { throw new PrintingException(400, "Invalid print options"); } @@ -982,6 +1135,7 @@ public class PrintingService } update(connection, "UPDATE printing_job SET status='FAILED', message=? WHERE id=?", CANCELED_BEFORE_DISPATCH, jobId); connection.commit(); + realtimeBus.publish(tenantId, deviceId); return (new JSONObject().put("accepted", true)); } catch(Exception exception) @@ -1067,6 +1221,41 @@ public class PrintingService } + /** Stream external documents without holding a database connection or buffering the entire body. */ + public java.io.InputStream openContent(String deviceId, String jobId) throws SQLException + { + JSONObject row; + try(Connection connection = dataSource.getConnection()) + { + requireDevice(connection, deviceId); + row = deviceJob(connection, deviceId, jobId, true); + if("QUEUED".equals(row.getString("status"))) + { + throw new PrintingException(409, "Claim the job before downloading"); + } + if(row.getInt("content_available") == 0) + { + throw new PrintingException(410, "Print document expired under the host retention policy"); + } + } + if(row.getLong("content_length") > realtimePolicy().documentLimit(row.getString("format"))) + { + throw new PrintingException(413, "Document exceeds the current configured format limit"); + } + java.io.InputStream input; + if(row.isNull("content_reference")) + { + input = new java.io.ByteArrayInputStream(Base64.getDecoder().decode(row.getString("content_base64"))); + } + else + { + verifyStoredReference(row); + input = documentStorage.open(row.getString("content_storage_table"), row.getString("content_reference")); + } + return (new PrintingDocumentStream(input, row.getLong("content_length"), row.getString("content_hash"))); + } + + /*************************************************************************** ** Record a replay-safe lifecycle event without reprinting ambiguous jobs. @@ -1120,6 +1309,7 @@ public class PrintingService } update(connection, "UPDATE printing_job SET status=?, message=?, spooler_job_id=? WHERE id=?", status, message, spooler == null ? row.optString("spooler_job_id") : spooler, jobId); connection.commit(); + realtimeBus.publish(tenantId, deviceId); } catch(Exception exception) { @@ -1404,7 +1594,7 @@ public class PrintingService String cutoff = Instant.now().minusSeconds(row.getInt("retention_days") * 86400L).toString(); long backlog = query(connection, "SELECT COUNT(*) AS total FROM printing_job WHERE tenant_id=? AND content_available=1 AND status IN ('SPOOLED','SPOOLER_COMPLETED') AND completed_at result = PrintingAgentAuthorization.claim(device); + case "Claim" -> + { + service.requireRealtimeReady(device); + result = PrintingAgentAuthorization.claim(device); + } case "Queued" -> { PrintingControlAuthorization.table(PrintingTables.JOB); @@ -204,7 +208,7 @@ public abstract class PrintingUiApiSpec extends AbstractJsonEndpointSpec case "Content" -> { PrintingControlAuthorization.job(device, job); - return (service.content(device, job)); + return (service.openContent(device, job)); } case "Event" -> { @@ -275,7 +279,7 @@ public abstract class PrintingUiApiSpec extends AbstractJsonEndpointSpec /** Receipt and history access apply child-record security independently of device access. */ - private static void command(String device, String id) throws Exception + static void command(String device, String id) throws Exception { QRecord command = PrintingControlAuthorization.record(PrintingTables.COMMAND, id); if(!device.equals(command.getValueString("device_id"))) diff --git a/src/main/resources/openapi/qqq-v1/printing.yaml b/src/main/resources/openapi/qqq-v1/printing.yaml index 88032c2..b7445d4 100644 --- a/src/main/resources/openapi/qqq-v1/printing.yaml +++ b/src/main/resources/openapi/qqq-v1/printing.yaml @@ -604,9 +604,10 @@ components: info: contact: email: "contact@kingsrook.com" - description: "Native current-user printing with secondary device proof and device-control-v1." + description: "Native current-user printing with secondary device proof, device-control-v1\ + \ and realtime-v1." title: "QQQ Middleware API extension: printing" - version: "1.0.0" + version: "1.1.0" openapi: "3.0.3" paths: /qqq/v1/qbit/printing/session: diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingAdmissionTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingAdmissionTest.java index 17424c1..8c4e666 100644 --- a/src/test/java/com/kingsrook/qbits/printing/PrintingAdmissionTest.java +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingAdmissionTest.java @@ -160,13 +160,13 @@ class PrintingAdmissionTest /*************************************************************************** - ** Legacy upload intents with unknown size reserve the maximum allowed bytes. + ** Legacy unknown sizes block new admission even when current capacity exceeds their historical estimate. ***************************************************************************/ @Test void legacyUnknownUploadChargesTenantCapacityUntilExactCleanup() throws Exception { long unknownBytes = 20L * 1024 * 1024; - Fixture fixture = fixture(limits(10, 20, 100_000_000, 200_000_000, 100_000_000, unknownBytes, 100, 200, 10, 20)); + Fixture fixture = fixture(limits(10, 20, 100_000_000, 200_000_000, 100_000_000, 200_000_000, 100, 200, 10, 20)); String orphan = UUID.randomUUID().toString(); String reference = fixture.documents.prefix() + orphan + ".zpl"; fixture.documents.objects.put(reference, DOCUMENT.clone()); @@ -572,7 +572,7 @@ class PrintingAdmissionTest ** Read immutable synthetic bytes for the retry path. ***********************************************************************/ @Override - public byte[] get(String table, String reference) throws SQLException + public byte[] get(String table, String reference, int maximum) throws SQLException { gets.incrementAndGet(); byte[] content = objects.get(reference); diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingDocumentStorageTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingDocumentStorageTest.java index adfbfa1..eacbf68 100644 --- a/src/test/java/com/kingsrook/qbits/printing/PrintingDocumentStorageTest.java +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingDocumentStorageTest.java @@ -255,6 +255,40 @@ class PrintingDocumentStorageTest } + /** Native StorageAction content larger than the old cap is consumed incrementally and verified. */ + @Test + void streamsLargePdfAndRejectsChangedLengthWithoutBufferingInTheAdapter() throws Exception + { + Fixture fixture = fixture(); + fixture.service.inventory(fixture.device, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic") + .put("formats", new JSONArray(List.of("PDF"))))); + byte[] document = new byte[21 * 1024 * 1024]; + System.arraycopy("%PDF-1.7\n".getBytes(java.nio.charset.StandardCharsets.US_ASCII), 0, document, 0, 9); + JSONObject submitted = fixture.service.submit(new JSONObject().put("deviceId", fixture.device).put("printerId", "queue").put("title", "Large synthetic PDF") + .put("format", "PDF").put("contentBase64", Base64.getEncoder().encodeToString(document)).put("idempotencyKey", "large-pdf")); + fixture.service.claim(fixture.device); + java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256"); + long count = 0; + try(var stream = fixture.service.openContent(fixture.device, submitted.getString("jobId"))) + { + byte[] buffer = new byte[65536]; + int read; + while((read = stream.read(buffer)) != -1) + { + count += read; + digest.update(buffer, 0, read); + } + } + assertEquals(document.length, count); + assertEquals(submitted.getString("contentSha256"), java.util.HexFormat.of().formatHex(digest.digest())); + fixture.storage.put("printing_documents", fixture.storage.lastReference, "PDF", new byte[] {1, 2, 3}); + try(var stream = fixture.service.openContent(fixture.device, submitted.getString("jobId"))) + { + assertThrows(java.io.IOException.class, stream::readAllBytes); + } + } + + /*************************************************************************** ** Synthetic bytes never leave the test process. diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeProcessesTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeProcessesTest.java new file mode 100644 index 0000000..1e5e33a --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeProcessesTest.java @@ -0,0 +1,201 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.math.BigDecimal; +import java.util.UUID; +import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput; +import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType; +import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData; +import com.kingsrook.qqq.backend.core.model.session.QSession; +import com.kingsrook.qqq.backend.core.model.session.QUser; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +/******************************************************************************* + ** Native settings forms preserve exact values and enforce current authority. + *******************************************************************************/ +class PrintingRealtimeProcessesTest +{ + /*************************************************************************** + ** Opening settings cannot save them until the operator passes native review. + ***************************************************************************/ + @Test + void nativeReviewIsReadOnly() throws Exception + { + PrintingService service = fixture(); + var preview = new RunProcessAction().execute(new RunProcessInput().withProcessName(PrintingRealtimeProcesses.SETTINGS)); + assertFalse(preview.getException().isPresent(), () -> preview.getException().toString()); + assertEquals("review", preview.getProcessState().getNextStepName().orElseThrow()); + assertEquals(0, service.getRealtimeSettings().getLong("revision")); + } + + + + /*************************************************************************** + ** Large PDFs and fractional human units round-trip without a fixed 20 MiB cap. + ***************************************************************************/ + @Test + void exactPolicyPersistsAndRejectsPartialOrInconsistentEdits() throws Exception + { + PrintingService service = fixture(); + RunBackendStepInput save = prepared(PrintingRealtimeProcesses.SETTINGS); + save.addValue("maxPdfMiB", new BigDecimal("250.5")); + save.addValue("maxConcurrentDownloads", 3); + new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput()); + assertEquals(262668288L, service.getRealtimeSettings().getLong("maxPdfBytes")); + assertEquals(3, service.getRealtimeSettings().getInt("maxConcurrentDownloads")); + assertEquals(1, service.getRealtimeSettings().getLong("revision")); + RunBackendStepInput fractionalByte = prepared(PrintingRealtimeProcesses.SETTINGS).addValue("maxPdfMiB", new BigDecimal("0.1")); + assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(fractionalByte, new RunBackendStepOutput())); + RunBackendStepInput tooLittleDisk = prepared(PrintingRealtimeProcesses.SETTINGS).addValue("maxStagedMiB", 1); + assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(tooLittleDisk, new RunBackendStepOutput())); + assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(input(PrintingRealtimeProcesses.SETTINGS, "execute"), new RunBackendStepOutput())); + assertEquals(1, service.getRealtimeSettings().getLong("revision")); + } + + + + /*************************************************************************** + ** Backend invocation cannot bypass either tenant scope or administrator role. + ***************************************************************************/ + @Test + void currentUserAndTenantAreRequired() throws Exception + { + fixture(); + QInstance instance = QContext.getQInstance(); + RunBackendStepInput save = prepared(PrintingRealtimeProcesses.SETTINGS); + QContext.init(instance, session("foreign", true)); + assertThrows(QPermissionDeniedException.class, () -> new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput())); + QContext.init(instance, session("tenant-a", false)); + assertThrows(QPermissionDeniedException.class, () -> new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput())); + } + + + + /*************************************************************************** + ** Queue limits use the same typed native forms and survive new service objects. + ***************************************************************************/ + @Test + void admissionSettingsPersistAndRejectInvalidCapacity() throws Exception + { + PrintingService service = fixture(); + RunBackendStepInput save = prepared(PrintingRealtimeProcesses.ADMISSION).addValue("deviceJobsPerHour", 15000).addValue("deviceOutstandingMiB", new BigDecimal("10240.5")); + new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput()); + assertEquals(15000, service.getAdmissionSettings().getInt("deviceJobsPerHour")); + assertEquals(10737942528L, service.getAdmissionSettings().getLong("deviceOutstandingBytes")); + RunBackendStepInput zero = prepared(PrintingRealtimeProcesses.ADMISSION).addValue("devicePrinters", 0); + assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(zero, new RunBackendStepOutput())); + assertTrue(service.getAdmissionSettings().getInt("devicePrinters") > 0); + } + + + + /*************************************************************************** + ** Prepared form values exercise the same complete fields used by the UI. + ***************************************************************************/ + private static RunBackendStepInput prepared(String process) throws Exception + { + RunBackendStepOutput output = new RunBackendStepOutput(); + new PrintingRealtimeProcesses().run(input(process, "prepare"), output); + RunBackendStepInput save = input(process, "execute"); + output.getValues().forEach(save::addValue); + return (save); + } + + + + /*************************************************************************** + ** Each test owns a database and normal authenticated tenant session. + ***************************************************************************/ + private static PrintingService fixture() throws Exception + { + DriverDataSource source = new DriverDataSource("jdbc:h2:mem:realtime-ui-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", ""); + PrintingService service = new PrintingService(source, "tenant-a"); + service.initialize(); + QInstance instance = new QInstance(); + instance.addBackend(new QBackendMetaData().withName("memory").withBackendType(com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule.class)); + instance.addTable(new QTableMetaData().withName("fixture").withBackendName("memory").withPrimaryKeyField("id").withField(new QFieldMetaData("id", QFieldType.STRING))); + instance.addQBit(new QBitMetaData().withGroupId("com.kingsrook.qbits").withArtifactId("qbit-printing").withVersion("test").withNamespace("printing").withConfig(new PrintingQBitConfig().withService(service))); + QAppMetaData app = new QAppMetaData().withName("printing").withLabel("Printing"); + PrintingRealtimeProcesses.register(instance, app); + instance.addApp(app); + instance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("mock").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.MOCK)); + QContext.init(instance, session("tenant-a", true)); + return (service); + } + + + + /*************************************************************************** + ** Negative authorization cases retain a real user to test role and tenant. + ***************************************************************************/ + private static QSession session(String tenant, boolean administrator) + { + QSession session = new QSession().withUser(new QUser().withIdReference("operator")).withSecurityKeyValue("printing_tenant", tenant); + if(administrator) + { + session.withPermissions("printing_admin"); + } + return (session); + } + + + + /*************************************************************************** + ** Backend entrypoints use explicit process and step names. + ***************************************************************************/ + private static RunBackendStepInput input(String process, String step) + { + RunBackendStepInput input = new RunBackendStepInput().withProcessName(process); + input.setStepName(step); + return (input); + } + + + + /*************************************************************************** + ** Never leak current tenant authority into another test. + ***************************************************************************/ + @AfterEach + void cleanup() + { + QContext.clear(); + } +} diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeTest.java new file mode 100644 index 0000000..4afd67e --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeTest.java @@ -0,0 +1,291 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.time.Instant; +import java.util.Base64; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import com.kingsrook.qqq.middleware.javalin.QApplicationLauncher; +import com.kingsrook.qqq.middleware.javalin.QApplicationLauncherConfig; +import io.javalin.Javalin; +import org.json.JSONArray; +import org.json.JSONObject; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + + +/** Real native authentication and WebSocket lifecycle with durable synthetic jobs only. */ +public class PrintingRealtimeTest +{ + @TempDir + Path directory; + + + /** Commit wakeups, exact receipts, replacement and revocation all use actual QQQ sessions. */ + @Test + void authenticatedPushReconcilesDurableWorkWithoutHttpControlPolling() throws Exception + { + String jdbc = "jdbc:h2:file:" + directory.resolve("realtime").toString().replace('\\', '/'); + PrintingLocalHost host = new PrintingLocalHost(new PrintingLocalSettings(jdbc, "sa", "", "demo", 0, "synthetic-management-token-long-enough", "operator", "synthetic-realtime-password-1234")); + AtomicReference app = new AtomicReference<>(); + QApplicationLauncher launcher = QApplicationLauncher.run(host, new QApplicationLauncherConfig().withRegisterShutdownHook(false).withServerCustomizer(server -> + { + host.configure(server, false); + server.withJavalinConfigurationCustomizer(app::set); + })); + PrintingService service = launcher.getQInstance().getQBits().values().stream().map(value -> value.getConfig()).filter(PrintingQBitConfig.class::isInstance) + .map(PrintingQBitConfig.class::cast).findFirst().orElseThrow().getService(); + String base = "http://127.0.0.1:" + app.get().port(); + JSONObject enrolled = service.enrollForUser(service.createEnrollment("default", "Synthetic realtime").put("deviceName", "Synthetic realtime").put("agentVersion", "test"), "operator"); + String device = enrolled.getString("deviceId"); + String proof = enrolled.getString("deviceToken"); + service.inventory(device, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic queue").put("status", "READY").put("formats", new JSONArray().put("ZPL")))); + service.enablePrinter(device, "queue", true); + try(HttpClient http = HttpClient.newHttpClient()) + { + String cookie = login(http, base); + assertThrows(Exception.class, () -> PrintingSocketFixture.connect(http, base, "sessionUUID=invalid", device, proof)); + assertThrows(Exception.class, () -> PrintingSocketFixture.connect(http, base, cookie, device, "wrong-proof")); + try(PrintingSocketFixture first = PrintingSocketFixture.connect(http, base, cookie, device, proof)) + { + first.take("JOB_AVAILABLE"); + first.send("INVENTORY", "idle-inventory", new JSONObject().put("printers", new JSONArray().put(new JSONObject().put("printerId", "queue") + .put("name", "Synthetic queue").put("status", "READY").put("formats", new JSONArray().put("ZPL"))))); + assertEquals("idle-inventory", first.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId")); + org.junit.jupiter.api.Assertions.assertNull(first.received.poll(300, TimeUnit.MILLISECONDS), "Idle inventory must not create a self-sustaining job wakeup"); + sharedBusRoundTrip(launcher, service, jdbc, http, cookie, device, first); + first.received.clear(); + String job = service.submit(request(device, "one")).getString("jobId"); + first.take("JOB_AVAILABLE"); + assertEquals(job, service.claim(device).getJSONObject("job").getString("jobId")); + JSONObject event = new JSONObject().put("eventId", "stable-event").put("status", "RECEIVED"); + first.send("JOB_EVENT", "event-stable-event", new JSONObject().put("jobId", job).put("event", event)); + assertEquals("event-stable-event", first.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId")); + assertEquals("RECEIVED", service.getJob(job).getString("status")); + JSONObject policy = service.getRealtimeSettings(); + policy.remove("revision"); + service.saveRealtimeSettings(policy.put("maxCopies", 99)); + assertEquals(1, first.take("SETTINGS").getJSONObject("payload").getLong("revision")); + first.apply(0, 0); + assertEquals(409, assertThrows(PrintingException.class, () -> service.requireRealtimeReady(device)).getStatus()); + first.apply(0, 1); + service.requireRealtimeReady(device); + JSONObject pause = service.deviceAction(device, action("PAUSE_DISPATCH"), "operator"); + JSONObject desired = first.take("DESIRED_STATE").getJSONObject("payload"); + while(!desired.getBoolean("dispatchPaused")) + { + desired = first.take("DESIRED_STATE").getJSONObject("payload"); + } + first.apply(desired.getLong("revision"), 1); + assertTrue(service.claim(device).isNull("job")); + assertFalse(pause.isEmpty()); + JSONObject command = service.deviceAction(device, action("REFRESH_INVENTORY"), "operator"); + JSONObject pushed = first.take("COMMAND").getJSONObject("payload"); + assertEquals(command.getString("commandId"), pushed.getString("commandId")); + JSONObject receipt = new JSONObject().put("receiptId", "stable-receipt").put("outcome", "INVENTORY_REFRESHED").put("completedAt", Instant.now().toString()); + JSONObject delivery = new JSONObject().put("commandId", pushed.getString("commandId")).put("receipt", receipt); + first.send("COMMAND_RECEIPT", "receipt-stable-receipt", delivery); + first.take("ACK"); + first.send("COMMAND_RECEIPT", "receipt-pruned", new JSONObject().put("commandId", UUID.randomUUID().toString()).put("receipt", receipt)); + assertEquals("NOT_FOUND", first.take("ERROR").getJSONObject("payload").getString("code")); + first.send("PING", "after-pruned-receipt", new JSONObject()); + first.take("PONG"); + try(PrintingSocketFixture second = PrintingSocketFixture.connect(http, base, cookie, device, proof)) + { + assertEquals(1008, first.closed.get(10, TimeUnit.SECONDS)); + second.send("COMMAND_RECEIPT", "receipt-stable-receipt", delivery); + assertEquals("receipt-stable-receipt", second.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId")); + second.send("JOB_EVENT", "event-stable-event", new JSONObject().put("jobId", job).put("event", event)); + second.take("ACK"); + assertEquals("RECEIVED", service.getJob(job).getString("status")); + service.deviceAction(device, action("REVOKE"), "operator"); + assertEquals(1008, second.closed.get(10, TimeUnit.SECONDS)); + } + } + authenticationChangesAreImmediate(launcher, service, http, base, cookie); + } + finally + { + launcher.stop(); + } + } + + + /** Cookie variants, changed provider roles, channel policy and logout are checked after upgrade. */ + private static void authenticationChangesAreImmediate(QApplicationLauncher launcher, PrintingService service, HttpClient http, String base, String cookie) throws Exception + { + JSONObject enrolled = service.enrollForUser(service.createEnrollment("default", "Authentication fixture").put("deviceName", "Authentication fixture").put("agentVersion", "test"), "operator"); + String device = enrolled.getString("deviceId"); + String proof = enrolled.getString("deviceToken"); + var table = launcher.getQInstance().getTable(PrintingTables.DEVICE); + var originalRules = table.getPermissionRules(); + try(PrintingSocketFixture channel = PrintingSocketFixture.connect(http, base, cookie.replace("sessionUUID=", "sessionId="), device, proof)) + { + table.withPermissionRules(PrintingProcesses.permissionRules().withDeniedClients(java.util.Set.of("api"))); + channel.send("PING", "channel-denied", new JSONObject()); + assertEquals("AUTHORIZATION_REQUIRED", channel.take("ERROR").getJSONObject("payload").getString("code")); + assertEquals(1008, channel.closed.get(10, TimeUnit.SECONDS)); + } + finally + { + table.withPermissionRules(originalRules); + } + var auth = launcher.getQInstance().getAuthentication(); + var originalModule = auth.getModuleCodeReference(); + try(PrintingSocketFixture channel = PrintingSocketFixture.connect(http, base, cookie, device, proof)) + { + auth.withModuleCodeReference(new com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference(RemovedRoleAuthentication.class)); + service.getRealtimeBus().publish("demo", device); + assertEquals("AUTHORIZATION_REQUIRED", channel.take("ERROR").getJSONObject("payload").getString("code")); + assertEquals(1008, channel.closed.get(10, TimeUnit.SECONDS)); + } + finally + { + auth.withModuleCodeReference(originalModule); + } + try(PrintingSocketFixture channel = PrintingSocketFixture.connect(http, base, cookie, device, proof)) + { + HttpResponse logout = http.send(HttpRequest.newBuilder(URI.create(base + "/qqq/v1/logout")).header("Origin", base).header("Cookie", cookie) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString("{}")).build(), HttpResponse.BodyHandlers.ofString()); + assertEquals(200, logout.statusCode()); + channel.send("PING", "logged-out", new JSONObject()); + assertEquals("AUTHORIZATION_REQUIRED", channel.take("ERROR").getJSONObject("payload").getString("code")); + assertEquals(1008, channel.closed.get(10, TimeUnit.SECONDS)); + } + } + + + /** Native authentication still verifies the actual password/session; only current role membership changes. */ + public static class RemovedRoleAuthentication extends PrintingLocalAuthentication + { + @Override + public com.kingsrook.qqq.backend.core.model.session.QSession createSession(com.kingsrook.qqq.backend.core.model.metadata.QInstance instance, java.util.Map context) + throws com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException + { + var session = super.createSession(instance, context); + session.setPermissions(java.util.Set.of()); + return (session); + } + } + + + /** Two independent route providers receive only their scoped commit wakeups through an explicitly shared bus. */ + private static void sharedBusRoundTrip(QApplicationLauncher launcher, PrintingService service, String jdbc, HttpClient http, String cookie, String firstDevice, PrintingSocketFixture first) throws Exception + { + PrintingService sibling = new PrintingService(new DriverDataSource(jdbc, "sa", ""), "demo").withRealtimeBus(service.getRealtimeBus()); + PrintingRealtimeRoutes routes = new PrintingRealtimeRoutes(sibling); + routes.setQInstance(launcher.getQInstance()); + Javalin other = Javalin.create(config -> + { + config.jetty.host = "127.0.0.1"; + routes.acceptJavalinConfig(config); + }).start(0); + JSONObject enrolled = service.enrollForUser(service.createEnrollment("default", "Second synthetic device").put("deviceName", "Second synthetic device").put("agentVersion", "test"), "operator"); + String secondDevice = enrolled.getString("deviceId"); + service.inventory(secondDevice, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic queue").put("formats", new JSONArray().put("ZPL")))); + service.enablePrinter(secondDevice, "queue", true); + try(PrintingSocketFixture second = PrintingSocketFixture.open(http, "http://127.0.0.1:" + other.port(), cookie, secondDevice, enrolled.getString("deviceToken"))) + { + String early = service.submit(request(secondDevice, "commit-between-upgrade-and-hello")).getString("jobId"); + second.initialize(); + second.take("JOB_AVAILABLE"); + assertEquals("QUEUED", sibling.getJob(early).getString("status")); + sibling.cancel(secondDevice, early); + first.received.clear(); + second.received.clear(); + String firstJob = sibling.submit(request(firstDevice, "other-service-to-first-hub")).getString("jobId"); + first.take("JOB_AVAILABLE"); + String secondJob = service.submit(request(secondDevice, "first-service-to-other-hub")).getString("jobId"); + second.take("JOB_AVAILABLE"); + assertEquals("QUEUED", sibling.getJob(firstJob).getString("status")); + assertEquals("QUEUED", service.getJob(secondJob).getString("status")); + service.cancel(firstDevice, firstJob); + sibling.cancel(secondDevice, secondJob); + } + finally + { + other.stop(); + } + } + + + /** A real browser login supplies only its own temporary test session. */ + private static String login(HttpClient client, String base) throws Exception + { + HttpRequest request = HttpRequest.newBuilder(URI.create(base + "/qqq/v1/manageSession")).header("Origin", base) + .header("Authorization", "Basic " + Base64.getEncoder().encodeToString("operator:synthetic-realtime-password-1234".getBytes(StandardCharsets.UTF_8))) + .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString("{}")).build(); + HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofString()); + assertEquals(200, response.statusCode()); + return (response.headers().allValues("set-cookie").stream().filter(value -> value.startsWith("sessionUUID=")).findFirst().orElseThrow().split(";", 2)[0]); + } + + + /** Stable synthetic submission identity; no physical printer exists in this fixture. */ + private static JSONObject request(String device, String key) + { + return (new JSONObject().put("deviceId", device).put("printerId", "queue").put("title", "Synthetic realtime") + .put("format", "ZPL").put("contentBase64", Base64.getEncoder().encodeToString("^XA^XZ".getBytes(StandardCharsets.US_ASCII))).put("idempotencyKey", key)); + } + + + /** Native actions retain their existing temporal replay boundary. */ + private static JSONObject action(String name) + { + return (new JSONObject().put("action", name).put("requestId", UUID.randomUUID().toString()).put("issuedAt", Instant.now().toString())); + } + + + /** Policy updates preserve validation, tenant isolation and durable revisions. */ + @Test + void policyRejectsInconsistentLimitsAndPersistsRevisions() throws Exception + { + DriverDataSource source = new DriverDataSource("jdbc:h2:mem:policy-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", ""); + PrintingService first = new PrintingService(source, "one"); + first.initialize(); + PrintingService other = new PrintingService(source, "two"); + JSONObject values = first.getRealtimeSettings(); + assertEquals(104857600, values.getInt("maxPdfBytes")); + values.remove("revision"); + JSONObject invalid = new JSONObject(values.toString()).put("heartbeatTimeoutSeconds", 10); + assertThrows(PrintingException.class, () -> first.saveRealtimeSettings(invalid)); + assertThrows(PrintingException.class, () -> first.saveRealtimeSettings(new JSONObject(values.toString()).put("maxCopies", 1.5))); + values.put("maxPdfBytes", 30 * 1024 * 1024); + assertEquals(1, first.saveRealtimeSettings(values).getLong("revision")); + assertEquals(1, new PrintingService(source, "one").getRealtimeSettings().getLong("revision")); + assertNotEquals(first.getRealtimeSettings().getLong("maxPdfBytes"), other.getRealtimeSettings().getLong("maxPdfBytes")); + assertThrows(PrintingException.class, () -> PrintingRealtimePolicy.defaults().requireClientLimits(new JSONObject().put("maxPdfBytes", 1))); + } +} diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingRetentionProcessesTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingRetentionProcessesTest.java index 8ff5e9f..2ca990b 100644 --- a/src/test/java/com/kingsrook/qbits/printing/PrintingRetentionProcessesTest.java +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingRetentionProcessesTest.java @@ -105,7 +105,7 @@ class PrintingRetentionProcessesTest { PrintingService service = fixture(); PrintingRetentionProcesses step = new PrintingRetentionProcesses(); - RunBackendStepInput save = input(PrintingRetentionProcesses.SETTINGS, "execute").addValue("days", 730).addValue("enabled", false); + RunBackendStepInput save = input(PrintingRetentionProcesses.SETTINGS, "execute").addValue("days", 730).addValue("enabled", false).addValue("batchSize", 100); QContext.getQSession().setSecurityKeyValues(Map.of("printing_tenant", List.of("foreign"))); assertThrows(QPermissionDeniedException.class, () -> step.run(save, new RunBackendStepOutput())); QContext.init(QContext.getQInstance(), new QSession().withSecurityKeyValue("printing_tenant", "tenant-a")); diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingSocketFixture.java b/src/test/java/com/kingsrook/qbits/printing/PrintingSocketFixture.java new file mode 100644 index 0000000..f40c7f3 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingSocketFixture.java @@ -0,0 +1,157 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.printing; + + +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.WebSocket; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import org.json.JSONArray; +import org.json.JSONObject; +import static org.junit.jupiter.api.Assertions.assertNotNull; + + +/** Real native WebSocket client shared by the HTTP and realtime integration fixtures. */ +final class PrintingSocketFixture implements WebSocket.Listener, AutoCloseable +{ + final BlockingQueue received = new LinkedBlockingQueue<>(); + final CompletableFuture closed = new CompletableFuture<>(); + private final StringBuilder text = new StringBuilder(); + private WebSocket socket; + + + /** Authenticate with the normal session cookie and secondary proof headers. */ + static PrintingSocketFixture connect(HttpClient client, String base, String cookie, String device, String proof) throws Exception + { + PrintingSocketFixture fixture = open(client, base, cookie, device, proof); + fixture.initialize(); + return (fixture); + } + + + /** Separate upgrade from HELLO so a commit in that interval can be tested deterministically. */ + static PrintingSocketFixture open(HttpClient client, String base, String cookie, String device, String proof) throws Exception + { + PrintingSocketFixture fixture = new PrintingSocketFixture(); + fixture.socket = client.newWebSocketBuilder().subprotocols(PrintingRealtimeRoutes.SUBPROTOCOL).header("Origin", base) + .header("X-QQQ-Client", "api/qqq-print-agent") + .header("Cookie", cookie).header(PrintingAgentAuthorization.DEVICE_HEADER, device).header(PrintingAgentAuthorization.TOKEN_HEADER, proof) + .buildAsync(URI.create(base.replace("http:", "ws:") + PrintingRealtimeRoutes.PATH), fixture).get(10, TimeUnit.SECONDS); + return (fixture); + } + + + /** Reconcile after the connection's subscription is already registered. */ + void initialize() throws Exception + { + send("HELLO", "hello", new JSONObject().put("agentVersion", "synthetic-test").put("capabilities", new JSONArray().put("realtime-v1").put("REFRESH_INVENTORY").put("CANCEL_SPOOL")) + .put("lastAppliedRevision", 0).put("lastSettingsRevision", 0).put("limits", PrintingRealtimePolicy.defaults().toJson(0))); + JSONObject welcome = take("WELCOME").getJSONObject("payload"); + apply(welcome.getJSONObject("desiredState").getLong("revision"), welcome.getJSONObject("settings").getLong("revision")); + } + + + /** Observe application ACK before allowing a dependent HTTP claim. */ + void apply(long state, long settings) throws Exception + { + send("STATE_APPLIED", "applied", new JSONObject().put("lastAppliedRevision", state).put("lastSettingsRevision", settings)); + take("ACK"); + } + + + /** Send one explicit protocol operation, never a timed query. */ + void send(String type, String id, JSONObject payload) + { + socket.sendText(new JSONObject().put("type", type).put("messageId", id).put("payload", payload).toString(), true).join(); + } + + + /** Bounded waiting diagnoses absent pushes without introducing a polling loop. */ + JSONObject take(String type) throws Exception + { + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while(System.nanoTime() < end) + { + JSONObject next = received.poll(Math.max(1, end - System.nanoTime()), TimeUnit.NANOSECONDS); + assertNotNull(next, "Expected realtime frame " + type); + if("ERROR".equals(next.getString("type")) && !"ERROR".equals(type)) + { + throw new AssertionError("Realtime rejected synthetic test message: " + next.getJSONObject("payload")); + } + if(type.equals(next.getString("type"))) + { + return (next); + } + } + throw new AssertionError("No frame " + type); + } + + + @Override + public void onOpen(WebSocket webSocket) + { + webSocket.request(1); + } + + + @Override + public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean last) + { + text.append(data); + if(last) + { + received.add(new JSONObject(text.toString())); + text.setLength(0); + } + webSocket.request(1); + return (null); + } + + + @Override + public CompletionStage onClose(WebSocket webSocket, int statusCode, String reason) + { + closed.complete(statusCode); + return (null); + } + + + @Override + public void onError(WebSocket webSocket, Throwable error) + { + closed.completeExceptionally(error); + } + + + @Override + public void close() + { + if(socket != null) + { + socket.abort(); + } + } +} diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingStorageBindingTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingStorageBindingTest.java index 5aed4ae..6c70741 100644 --- a/src/test/java/com/kingsrook/qbits/printing/PrintingStorageBindingTest.java +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingStorageBindingTest.java @@ -405,7 +405,7 @@ class PrintingStorageBindingTest ** Unexpected reads are observable independently of returned bytes. ************************************************************************/ @Override - public byte[] get(String table, String reference) + public byte[] get(String table, String reference, int maximum) { calls.add("GET " + reference); return (objects.get(reference).clone()); diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiContractTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiContractTest.java index a27ef28..2f2613f 100644 --- a/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiContractTest.java +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiContractTest.java @@ -61,7 +61,7 @@ class PrintingUiApiContractTest var model = version.generateExtensionOpenAPIModel("qqq", extension); assertEquals(17, model.getPaths().size()); assertTrue(model.getPaths().keySet().stream().allMatch(path -> path.startsWith("/qqq/v1/qbit/printing/"))); - assertEquals("1.0.0", extension.getSemanticVersion()); + assertEquals("1.1.0", extension.getSemanticVersion()); assertTrue(generated.contains("X-QQQ-Device-Token")); assertTrue(generated.contains("application/octet-stream")); assertTrue(generated.contains("principalId")); diff --git a/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiTest.java b/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiTest.java index fd7c3ff..b09589b 100644 --- a/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiTest.java +++ b/src/test/java/com/kingsrook/qbits/printing/PrintingUiApiTest.java @@ -136,6 +136,8 @@ class PrintingUiApiTest JSONObject others = service.enrollForUser(service.createEnrollment("default", "Other owner").put("deviceName", "Other owner").put("agentVersion", "test"), "another-principal"); assertEquals(403, send(client, base, prefix + "/device/heartbeat", "POST", new JSONObject().put("agentVersion", "test"), cookie, others.getString("deviceId"), others.getString("deviceToken")).statusCode()); assertEquals(device, json(send(client, base, prefix + "/session", "GET", null, cookie, device, proof), 200).getString("deviceId")); + assertEquals(409, send(client, base, prefix + "/device/jobs/claim", "POST", new JSONObject(), cookie, device, proof).statusCode()); + PrintingSocketFixture channel = PrintingSocketFixture.connect(client, base, cookie, device, proof); assertEquals(job, json(send(client, base, prefix + "/device/jobs/claim", "POST", new JSONObject(), cookie, device, proof), 200).getJSONObject("job").getString("jobId")); HttpResponse downloaded = send(client, base, prefix + "/device/jobs/" + job + "/content", "GET", null, cookie, device, proof); assertEquals(200, downloaded.statusCode()); @@ -146,6 +148,7 @@ class PrintingUiApiTest JSONObject action = action("REVOKE"); json(send(client, base, prefix + "/admin/devices/" + device + "/actions", "POST", action, cookie, null, null), 200); assertEquals(403, send(client, base, prefix + "/device/heartbeat", "POST", new JSONObject().put("agentVersion", "test"), cookie, device, proof).statusCode()); + channel.close(); JSONObject nativeEnrollment = service.createEnrollment("default", "Native enrollment").put("deviceName", "Native enrollment").put("agentVersion", "test"); JSONObject fresh = json(send(client, base, prefix + "/devices/enroll", "POST", nativeEnrollment, cookie, null, null), 200); service.authorizeAgent(fresh.getString("deviceId"), fresh.getString("deviceToken"), "operator", false);