diff --git a/Documentation/client-snippets/contributing/clients/typescript-grpc-package/namespaces-definition.md b/Documentation/client-snippets/contributing/clients/typescript-grpc-package/namespaces-definition.md index 58a79e1..1873287 100644 --- a/Documentation/client-snippets/contributing/clients/typescript-grpc-package/namespaces-definition.md +++ b/Documentation/client-snippets/contributing/clients/typescript-grpc-package/namespaces-definition.md @@ -1,7 +1,7 @@ ```typescript import { NamespacesDefinition } from '@cratis/chronicle.contracts'; -const getNamespacesMethod = NamespacesDefinition.methods.getNamespaces; +const allNamespacesMethod = NamespacesDefinition.methods.allNamespaces; -console.log(`${NamespacesDefinition.fullName}.${getNamespacesMethod.name}`); +console.log(`${NamespacesDefinition.fullName}.${allNamespacesMethod.name}`); ``` diff --git a/Documentation/client-snippets/contributing/clients/typescript-grpc-package/request-messages.md b/Documentation/client-snippets/contributing/clients/typescript-grpc-package/request-messages.md index a8bf7aa..eda2604 100644 --- a/Documentation/client-snippets/contributing/clients/typescript-grpc-package/request-messages.md +++ b/Documentation/client-snippets/contributing/clients/typescript-grpc-package/request-messages.md @@ -1,9 +1,9 @@ ```typescript -import { EnsureEventStore, EnsureNamespace, GetNamespacesRequest } from '@cratis/chronicle.contracts'; +import { EnsureEventStoreRequest, EnsureNamespaceRequest, AllNamespacesRequest } from '@cratis/chronicle.contracts'; -const eventStore = EnsureEventStore.create({ Name: 'shopping' }); -const namespace = EnsureNamespace.create({ EventStore: eventStore.Name, Name: 'tenant-one' }); -const namespaces = GetNamespacesRequest.create({ EventStore: eventStore.Name }); +const eventStore = EnsureEventStoreRequest.create({ Name: 'shopping' }); +const namespace = EnsureNamespaceRequest.create({ EventStore: eventStore.Name, Namespace: 'tenant-one' }); +const namespaces = AllNamespacesRequest.create({ EventStore: eventStore.Name }); console.log(eventStore, namespace, namespaces); ``` diff --git a/Documentation/client-snippets/contributing/clients/typescript-grpc-package/service-types.md b/Documentation/client-snippets/contributing/clients/typescript-grpc-package/service-types.md index 50afc42..933076f 100644 --- a/Documentation/client-snippets/contributing/clients/typescript-grpc-package/service-types.md +++ b/Documentation/client-snippets/contributing/clients/typescript-grpc-package/service-types.md @@ -2,9 +2,13 @@ import { EventStoresClient, NamespacesClient } from '@cratis/chronicle.contracts'; async function readAvailableNamespaces(eventStores: EventStoresClient, namespaces: NamespacesClient): Promise { - await eventStores.ensure({ Name: 'shopping' }); - const result = await namespaces.getNamespaces({ EventStore: 'shopping' }); + await eventStores.ensureEventStore({ Name: 'shopping' }); - return result.items; + // Queries stream results so they can also be observed; one-shot callers take the first result. + for await (const result of namespaces.allNamespaces({ EventStore: 'shopping' })) { + return result.Data; + } + + return []; } ``` diff --git a/Documentation/client-snippets/jobs/index/get-one.md b/Documentation/client-snippets/jobs/index/get-one.md index 8e9df11..770a9a6 100644 --- a/Documentation/client-snippets/jobs/index/get-one.md +++ b/Documentation/client-snippets/jobs/index/get-one.md @@ -1,11 +1,11 @@ ```typescript import { IEventStore } from '@cratis/chronicle'; -import { Job } from '@cratis/chronicle.contracts'; +import { JobSummaryResponse } from '@cratis/chronicle.contracts'; class JobsIndexGetOne { constructor(private readonly store: IEventStore) {} - async getJob(jobId: string): Promise { + async getJob(jobId: string): Promise { return this.store.jobs.getJob(jobId); } } diff --git a/Documentation/client-snippets/jobs/index/get-steps.md b/Documentation/client-snippets/jobs/index/get-steps.md index e779605..48359a0 100644 --- a/Documentation/client-snippets/jobs/index/get-steps.md +++ b/Documentation/client-snippets/jobs/index/get-steps.md @@ -1,11 +1,11 @@ ```typescript import { IEventStore } from '@cratis/chronicle'; -import { JobStep } from '@cratis/chronicle.contracts'; +import { JobStepSummaryResponse } from '@cratis/chronicle.contracts'; class JobsIndexGetSteps { constructor(private readonly store: IEventStore) {} - async getSteps(jobId: string): Promise { + async getSteps(jobId: string): Promise { return this.store.jobs.getJobSteps(jobId); } } diff --git a/Documentation/client-snippets/jobs/index/list-all.md b/Documentation/client-snippets/jobs/index/list-all.md index eb22882..0a55eef 100644 --- a/Documentation/client-snippets/jobs/index/list-all.md +++ b/Documentation/client-snippets/jobs/index/list-all.md @@ -1,11 +1,11 @@ ```typescript import { IEventStore } from '@cratis/chronicle'; -import { Job } from '@cratis/chronicle.contracts'; +import { JobSummaryResponse } from '@cratis/chronicle.contracts'; class JobsIndexListAll { constructor(private readonly store: IEventStore) {} - async getAllJobs(): Promise { + async getAllJobs(): Promise { return this.store.jobs.getJobs(); } } diff --git a/Source/ChronicleClient.ts b/Source/ChronicleClient.ts index c337bf1..4d65186 100644 --- a/Source/ChronicleClient.ts +++ b/Source/ChronicleClient.ts @@ -6,6 +6,7 @@ import { diag } from '@opentelemetry/api'; import { SpanStatusCode } from '@opentelemetry/api'; import { ChronicleOptions } from './ChronicleOptions'; import { ChronicleConnection } from './connection'; +import { ensureCommandSuccess, ensureQuerySuccess, firstQueryResult } from './connection/callResults'; import { ConnectionLifecycle } from './connection/ConnectionLifecycle'; import { KernelKeepAlive } from './connection/KernelKeepAlive'; import { EventStore } from './EventStore'; @@ -131,7 +132,7 @@ export class ChronicleClient implements IChronicleClient { this._logger.debug('Ensuring event store exists in kernel', { eventStore: storeName.value }); - await this._connection.eventStores.ensure({ Name: storeName.value }); + ensureCommandSuccess('ensure event store', await this._connection.eventStores.ensureEventStore({ Name: storeName.value })); const created = new EventStore(storeName, namespaceName, this._connection, this._lifecycle, this.options.defaultSinkTypeId); this._stores.set(key, created); @@ -167,9 +168,9 @@ export class ChronicleClient implements IChronicleClient { try { const response = await this.withReconnect('get_event_stores', async () => { await this.ensureConnected(); - return this._connection.eventStores.getEventStores({}); + return firstQueryResult('get event stores', this._connection.eventStores.allEventStores({})); }); - const result = (response.items ?? []).map((name: string) => new EventStoreName(name)); + const result = ensureQuerySuccess('get event stores', response).map((name: string) => new EventStoreName(name)); this._logger.verbose('Retrieved event stores from kernel', { count: result.length }); diff --git a/Source/EventStore.ts b/Source/EventStore.ts index 2c1751c..6731c4f 100644 --- a/Source/EventStore.ts +++ b/Source/EventStore.ts @@ -5,6 +5,7 @@ import { diag } from '@opentelemetry/api'; import { SpanStatusCode } from '@opentelemetry/api'; import { ChronicleConnection } from './connection'; import { ConnectionLifecycle } from './connection/ConnectionLifecycle'; +import { ensureQuerySuccess, firstQueryResult } from './connection/callResults'; import { EventLog } from './eventSequences/EventLog'; import { EventSequence } from './eventSequences/EventSequence'; import { EventSequenceId } from './eventSequences/EventSequenceId'; @@ -163,8 +164,8 @@ export class EventStore implements IEventStore { return ChronicleTracer.startActiveSpan('chronicle.event_store.get_namespaces', async span => { span.setAttribute('chronicle.event_store', this.name.value); try { - const response = await this._connection.namespaces.getNamespaces({ EventStore: this.name.value }); - const result = (response.items ?? []).map((namespace: string) => new EventStoreNamespaceName(namespace)); + const response = await firstQueryResult('get namespaces', this._connection.namespaces.allNamespaces({ EventStore: this.name.value })); + const result = ensureQuerySuccess('get namespaces', response).map((namespace: string) => new EventStoreNamespaceName(namespace)); span.setStatus({ code: SpanStatusCode.OK }); return result; } catch (error) { diff --git a/Source/connection/callResults.spec.ts b/Source/connection/callResults.spec.ts new file mode 100644 index 0000000..5fe804e --- /dev/null +++ b/Source/connection/callResults.spec.ts @@ -0,0 +1,90 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import { describe, expect, it } from 'vitest'; +import { ChronicleCallFailed, ensureCommandSuccess, ensureQuerySuccess, firstQueryResult, isCallSuccess } from './callResults'; + +const successfulResult = { + ValidationResults: [], + ExceptionMessages: [], + AuthorizationFailureReason: '' +}; + +describe('callResults', () => { + describe('when checking a successful result', () => { + it('should report success', () => { + expect(isCallSuccess(successfulResult)).toBe(true); + }); + }); + + describe('when checking a result with an authorization failure reason', () => { + it('should report failure', () => { + expect(isCallSuccess({ ...successfulResult, AuthorizationFailureReason: 'no access' })).toBe(false); + }); + }); + + describe('when checking a result with validation results', () => { + it('should report failure', () => { + expect(isCallSuccess({ ...successfulResult, ValidationResults: [{ Message: 'invalid', Members: [] }] })).toBe(false); + }); + }); + + describe('when checking a result with exception messages', () => { + it('should report failure', () => { + expect(isCallSuccess({ ...successfulResult, ExceptionMessages: ['boom'] })).toBe(false); + }); + }); + + describe('when ensuring a successful command', () => { + it('should not throw', () => { + expect(() => ensureCommandSuccess('operation', successfulResult)).not.toThrow(); + }); + }); + + describe('when ensuring a failed command', () => { + it('should throw with the failure reasons', () => { + expect(() => ensureCommandSuccess('operation', { ...successfulResult, ExceptionMessages: ['boom'] })) + .toThrow(ChronicleCallFailed); + expect(() => ensureCommandSuccess('operation', { ...successfulResult, ExceptionMessages: ['boom'] })) + .toThrow(/operation.*boom/); + }); + }); + + describe('when ensuring a successful query', () => { + it('should return the data', () => { + expect(ensureQuerySuccess('operation', { ...successfulResult, Data: ['a', 'b'] })).toEqual(['a', 'b']); + }); + }); + + describe('when ensuring a failed query', () => { + it('should throw with the failure reasons', () => { + expect(() => ensureQuerySuccess('operation', { ...successfulResult, AuthorizationFailureReason: 'no access', Data: [] })) + .toThrow(/no access/); + }); + }); + + describe('when taking the first result from a stream', () => { + it('should return the first item and cancel the stream', async () => { + let cleanedUp = false; + async function* stream() { + try { + yield 'first'; + yield 'second'; + } finally { + cleanedUp = true; + } + } + + const result = await firstQueryResult('operation', stream()); + expect(result).toBe('first'); + expect(cleanedUp).toBe(true); + }); + }); + + describe('when taking the first result from an empty stream', () => { + it('should throw', async () => { + async function* stream() { /* produces nothing */ } + await expect(firstQueryResult('operation', stream())).rejects.toThrow(/completed without producing a result/); + }); + }); +}); diff --git a/Source/connection/callResults.ts b/Source/connection/callResults.ts new file mode 100644 index 0000000..20d4903 --- /dev/null +++ b/Source/connection/callResults.ts @@ -0,0 +1,112 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +/** + * Shape of a validation result carried by Chronicle command and query results. + */ +export interface CallValidationResult { + Message: string; + Members: string[]; +} + +/** + * Structural shape shared by every Chronicle command and query result envelope. + * + * The envelopes also carry an `IsAuthorized` boolean, but it is deliberately not part of + * this shape: the kernel declares it with a protobuf-net default of `true`, which means + * the field is omitted from the wire whenever it is true. A proto3 client decodes that + * absence as `false`, so the boolean always reads as `false` and carries no signal. + * Authorization failures are reported through `AuthorizationFailureReason` and gRPC + * status codes instead. + */ +export interface CallResultLike { + ValidationResults: CallValidationResult[]; + ExceptionMessages: string[]; + AuthorizationFailureReason?: string; +} + +/** + * Structural shape shared by every Chronicle query result envelope. + */ +export interface QueryResultLike extends CallResultLike { + Data: TData; +} + +/** + * Error thrown when a Chronicle command or query did not succeed. + */ +export class ChronicleCallFailed extends Error { + /** + * Creates a new {@link ChronicleCallFailed}. + * @param operation - The operation that failed. + * @param result - The result envelope returned by the kernel. + */ + constructor(operation: string, result: CallResultLike) { + const reasons: string[] = []; + if (result.AuthorizationFailureReason) { + reasons.push(result.AuthorizationFailureReason); + } + for (const validationResult of result.ValidationResults ?? []) { + reasons.push(validationResult.Message); + } + for (const exceptionMessage of result.ExceptionMessages ?? []) { + reasons.push(exceptionMessage); + } + + super(`Chronicle operation '${operation}' failed: ${reasons.join(', ') || 'unknown reason'}`); + this.name = 'ChronicleCallFailed'; + } +} + +/** + * Determines whether a Chronicle result envelope represents success: no authorization + * failure reason, no validation results, and no exceptions. + * @param result - The result envelope to check. + * @returns True when the call succeeded, false otherwise. + */ +export function isCallSuccess(result: CallResultLike): boolean { + return !result.AuthorizationFailureReason && + (result.ValidationResults ?? []).length === 0 && + (result.ExceptionMessages ?? []).length === 0; +} + +/** + * Ensures a Chronicle command executed successfully, throwing when it did not. + * @param operation - The operation the result belongs to, used for error reporting. + * @param result - The command result envelope returned by the kernel. + */ +export function ensureCommandSuccess(operation: string, result: CallResultLike): void { + if (!isCallSuccess(result)) { + throw new ChronicleCallFailed(operation, result); + } +} + +/** + * Ensures a Chronicle query executed successfully, returning its data or throwing when it did not. + * @param operation - The operation the result belongs to, used for error reporting. + * @param result - The query result envelope returned by the kernel. + * @returns The data produced by the query. + */ +export function ensureQuerySuccess(operation: string, result: QueryResultLike): TData { + if (!isCallSuccess(result)) { + throw new ChronicleCallFailed(operation, result); + } + + return result.Data; +} + +/** + * Takes the first result from a server-streaming Chronicle query and cancels the stream. + * Chronicle streams queries so they can also be observed; one-shot callers only need the + * first snapshot. + * @param operation - The operation the stream belongs to, used for error reporting. + * @param stream - The server-streaming query response. + * @returns The first result produced by the stream. + */ +export async function firstQueryResult(operation: string, stream: AsyncIterable): Promise { + for await (const result of stream) { + return result; + } + + throw new Error(`Chronicle operation '${operation}' completed without producing a result.`); +} diff --git a/Source/events/constraints/Constraints.ts b/Source/events/constraints/Constraints.ts index bc22887..7d44288 100644 --- a/Source/events/constraints/Constraints.ts +++ b/Source/events/constraints/Constraints.ts @@ -60,7 +60,7 @@ export class Constraints implements IConstraints { return { Name: capture.name, Type: ConstraintType.Unique, - RemovedWith: uc.removedWithEventTypeId ?? '', + RemovedWith: uc.removedWithEventTypeId ? [uc.removedWithEventTypeId] : [], Definition: { Value0: { EventDefinitions: uc.eventDefinitions.map(ed => ({ @@ -80,7 +80,7 @@ export class Constraints implements IConstraints { return { Name: uet.name ?? capture.name, Type: ConstraintType.UniqueEventType, - RemovedWith: '', + RemovedWith: [], Definition: { Value0: undefined, Value1: { @@ -94,7 +94,7 @@ export class Constraints implements IConstraints { return { Name: capture.name, Type: ConstraintType.Unknown, - RemovedWith: '', + RemovedWith: [], Definition: undefined, Scope: scope }; diff --git a/Source/jobs/IJobs.ts b/Source/jobs/IJobs.ts index 128ce04..242cba8 100644 --- a/Source/jobs/IJobs.ts +++ b/Source/jobs/IJobs.ts @@ -2,7 +2,7 @@ // Licensed under the MIT license. See LICENSE file in the project root for full license information. import { Guid } from '@cratis/fundamentals'; -import type { Job, JobStep } from '@cratis/chronicle.contracts'; +import type { JobSummaryResponse, JobStepSummaryResponse } from '@cratis/chronicle.contracts'; import { JobId } from './JobId'; /** @@ -32,18 +32,18 @@ export interface IJobs { * @param jobId - The job identifier. * @returns The job, or undefined when not found. */ - getJob(jobId: JobId | Guid | string): Promise; + getJob(jobId: JobId | Guid | string): Promise; /** * Gets all jobs for the event store namespace. * @returns All jobs. */ - getJobs(): Promise; + getJobs(): Promise; /** * Gets all steps for a specific job. * @param jobId - The job identifier. * @returns The job steps. */ - getJobSteps(jobId: JobId | Guid | string): Promise; + getJobSteps(jobId: JobId | Guid | string): Promise; } diff --git a/Source/jobs/Jobs.ts b/Source/jobs/Jobs.ts index 9b47cb5..a7fd39b 100644 --- a/Source/jobs/Jobs.ts +++ b/Source/jobs/Jobs.ts @@ -1,10 +1,11 @@ // Copyright (c) Cratis. All rights reserved. // Licensed under the MIT license. See LICENSE file in the project root for full license information. -import type { Job, JobStep } from '@cratis/chronicle.contracts'; +import type { JobSummaryResponse, JobStepSummaryResponse } from '@cratis/chronicle.contracts'; import { Guid } from '@cratis/fundamentals'; import { ChronicleConnection } from '../connection'; -import { toContractsGuid } from '../connection/Guid'; +import { ensureCommandSuccess, ensureQuerySuccess, firstQueryResult } from '../connection/callResults'; +import { fromContractsGuid, toContractsGuid } from '../connection/Guid'; import { IJobs } from './IJobs'; import { JobId } from './JobId'; @@ -26,39 +27,40 @@ export class Jobs implements IJobs { /** @inheritdoc */ async stop(jobId: JobId | Guid | string): Promise { - await this._connection.jobs.stop(this.createJobRequest(jobId)); + ensureCommandSuccess('stop job', await this._connection.jobs.stopJob(this.createJobRequest(jobId))); } /** @inheritdoc */ async resume(jobId: JobId | Guid | string): Promise { - await this._connection.jobs.resume(this.createJobRequest(jobId)); + ensureCommandSuccess('resume job', await this._connection.jobs.resumeJob(this.createJobRequest(jobId))); } /** @inheritdoc */ async delete(jobId: JobId | Guid | string): Promise { - await this._connection.jobs.delete(this.createJobRequest(jobId)); + ensureCommandSuccess('delete job', await this._connection.jobs.deleteJob(this.createJobRequest(jobId))); } /** @inheritdoc */ - async getJob(jobId: JobId | Guid | string): Promise { - const result = await this._connection.jobs.getJob(this.createJobRequest(jobId)); - return result.Value0; + async getJob(jobId: JobId | Guid | string): Promise { + const target = this.normalizeJobId(jobId).toString(); + const jobs = await this.getJobs(); + return jobs.find(job => fromContractsGuid(job.Id).toString() === target); } /** @inheritdoc */ - async getJobs(): Promise { - const response = await this._connection.jobs.getJobs({ + async getJobs(): Promise { + const response = await firstQueryResult('get jobs', this._connection.jobs.allJobs({ EventStore: this._eventStore, Namespace: this._namespace - }); + })); - return response.items ?? []; + return ensureQuerySuccess('get jobs', response); } /** @inheritdoc */ - async getJobSteps(jobId: JobId | Guid | string): Promise { + async getJobSteps(jobId: JobId | Guid | string): Promise { const response = await this._connection.jobs.getJobSteps(this.createJobRequest(jobId)); - return response.items ?? []; + return ensureQuerySuccess('get job steps', response); } private createJobRequest(jobId: JobId | Guid | string): { EventStore: string; Namespace: string; JobId: ReturnType } { diff --git a/Source/package.json b/Source/package.json index e7c0065..30cbaf2 100644 --- a/Source/package.json +++ b/Source/package.json @@ -160,7 +160,7 @@ }, "dependencies": { "@bufbuild/protobuf": "^2.12.0", - "@cratis/chronicle.contracts": "16.13.4", + "@cratis/chronicle.contracts": "17.0.0", "@grpc/grpc-js": "^1.14.4", "@opentelemetry/api": "^1.9.1", "nice-grpc": "^2.1.16", diff --git a/package.json b/package.json index 65ad946..1964396 100644 --- a/package.json +++ b/package.json @@ -14,7 +14,7 @@ "publish-version": "node ./publish-version.js" }, "dependencies": { - "@cratis/chronicle.contracts": "16.3.1", + "@cratis/chronicle.contracts": "17.0.0", "glob": "^13.0.6" } } diff --git a/yarn.lock b/yarn.lock index c2f69cf..6014cc7 100644 --- a/yarn.lock +++ b/yarn.lock @@ -33,25 +33,14 @@ __metadata: languageName: unknown linkType: soft -"@cratis/chronicle.contracts@npm:16.13.4": - version: 16.13.4 - resolution: "@cratis/chronicle.contracts@npm:16.13.4" +"@cratis/chronicle.contracts@npm:17.0.0": + version: 17.0.0 + resolution: "@cratis/chronicle.contracts@npm:17.0.0" dependencies: "@bufbuild/protobuf": "npm:^2.0.0" "@grpc/grpc-js": "npm:^1.12.4" nice-grpc: "npm:^2.1.0" - checksum: 10c0/427038a69b72a8871c330a23627c088c5f81a9d2a499157fb610304fbb9f18b40072f99b08bab22fb5617d9858bc151ff3c320e9c4bba0da6eba08a6b24eac3e - languageName: node - linkType: hard - -"@cratis/chronicle.contracts@npm:16.3.1": - version: 16.3.1 - resolution: "@cratis/chronicle.contracts@npm:16.3.1" - dependencies: - "@bufbuild/protobuf": "npm:^2.0.0" - "@grpc/grpc-js": "npm:^1.12.4" - nice-grpc: "npm:^2.1.0" - checksum: 10c0/734349b3dfc737ea1281e83e2091d9445ea9ce0a0a6547547be9dcfa4995067f5c71984944d271c684d90c710f970994076d9bf254720e4e0d7a410d5687cbf8 + checksum: 10c0/98af30515d6e77c8052c1143aee3206281987b0b20222ad7f338b2c37841ade217a2086983c72b6cf29b6207bd9a5fcd0ec6ba634ec0ad54d50f1999b08962c7 languageName: node linkType: hard @@ -60,7 +49,7 @@ __metadata: resolution: "@cratis/chronicle@workspace:Source" dependencies: "@bufbuild/protobuf": "npm:^2.12.0" - "@cratis/chronicle.contracts": "npm:16.13.4" + "@cratis/chronicle.contracts": "npm:17.0.0" "@cratis/fundamentals": "npm:7.14.0" "@grpc/grpc-js": "npm:^1.14.4" "@opentelemetry/api": "npm:^1.9.1" @@ -1929,7 +1918,7 @@ __metadata: version: 0.0.0-use.local resolution: "chronicle-typescript-workspace@workspace:." dependencies: - "@cratis/chronicle.contracts": "npm:16.3.1" + "@cratis/chronicle.contracts": "npm:17.0.0" glob: "npm:^13.0.6" languageName: unknown linkType: soft