Adding optional attributes in state management API (#476)

* Adding optional parameters concurrency, consistency and etag in state management

Signed-off-by: SoTrx <[email protected]>

* style(state): Changing one-liners ifs into multiline ifs and adding newlines

Signed-off-by: SoTrx <[email protected]>

* refactor(state): Renaming enums methods

matchStateConsistency -> getStateConsistencyValue
matchStateConcurrency -> getStateConcurrencyValue

Signed-off-by: SoTrx <[email protected]>

* fix(state): Querystring in get request was not used

Signed-off-by: SoTrx <[email protected]>

* refactor(state): Renaming StateConcurrency and StateConsistency enums to match other exported enums

Signed-off-by: SoTrx <[email protected]>

* chore(state): Adding StateConcurrencyEnum and StateConsistencyEnum as exported memeber for the whole lib

Signed-off-by: SoTrx <[email protected]>

* style : prettier

Signed-off-by: SoTrx <[email protected]>

* style : prettier

Signed-off-by: SoTrx <[email protected]>

* fix(state): Remapping enums values to their string representation when present in body

Signed-off-by: SoTrx <[email protected]>

* style(utils): Removing unused import

Signed-off-by: SoTrx <[email protected]>

---------

Signed-off-by: SoTrx <[email protected]>
Co-authored-by: SoTrx <[email protected]>

Author: SoTrx

src/enum/StateConcurrency.enum.ts

@@ -11,8 +11,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-export enum EStateConcurrency {
+export enum StateConcurrencyEnum {
   CONCURRENCY_UNSPECIFIED = 0,
   CONCURRENCY_FIRST_WRITE = 1,
   CONCURRENCY_LAST_WRITE = 2,
 }
+
+export default StateConcurrencyEnum;

src/enum/StateConsistency.enum.ts

@@ -11,8 +11,9 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-export enum EStateConsistency {
+export enum StateConsistencyEnum {
   CONSISTENCY_UNSPECIFIED = 0,
   CONSISTENCY_EVENTUAL = 1,
   CONSISTENCY_STRONG = 2,
 }
+export default StateConsistencyEnum;

src/implementation/Client/GRPCClient/state.ts

@@ -38,6 +38,9 @@ import { Settings } from "../../../utils/Settings.util";
 import { addMetadataToMap } from "../../../utils/Client.util";
 import { StateSaveResponseType } from "../../../types/state/StateSaveResponseType";
 import { StateSaveOptions } from "../../../types/state/StateSaveOptions.type";
+import { StateDeleteOptions } from "../../../types/state/StateDeleteOptions.type";
+import { StateGetOptions } from "../../../types/state/StateGetOptions.type";
+import { IStateOptions } from "../../../types/state/StateOptions.type";
 
 // https://docs.dapr.io/reference/api/state_api/
 export default class GRPCClientState implements IClientState {
@@ -63,6 +66,15 @@ export default class GRPCClientState implements IClientState {
           "utf-8",
         ),
       );
+
+      if (stateObject?.etag) {
+        const etag = new Etag();
+        etag.setValue(stateObject.etag);
+        si.setEtag(etag);
+      }
+
+      si.setOptions(this._configureStateOptions(stateObject?.options));
+
       // Merge metadata from stateObject and options.
       // Note, metadata from options will override metadata from stateObject.
       // See https://github.com/dapr/dapr/blob/029ec8cb7a1c88ec5d222bc2b0d1d53541217f19/pkg/http/api.go#L1525-L1532
@@ -89,13 +101,14 @@ export default class GRPCClientState implements IClientState {
     });
   }
 
-  async get(storeName: string, key: string): Promise<KeyValueType | string> {
+  async get(storeName: string, key: string, options?: Partial<StateGetOptions>): Promise<KeyValueType | string> {
     const msgService = new GetStateRequest();
     msgService.setStoreName(storeName);
     msgService.setKey(key);
 
-    // @todo: https://docs.dapr.io/reference/api/state_api/#optional-behaviors
-    // msgService.setConsistency()
+    if (options?.consistency) {
+      msgService.setConsistency(options.consistency as any);
+    }
 
     const client = await this.client.getClient();
 
@@ -158,14 +171,18 @@ export default class GRPCClientState implements IClientState {
     });
   }
 
-  async delete(storeName: string, key: string): Promise<void> {
+  async delete(storeName: string, key: string, options?: StateDeleteOptions): Promise<StateSaveResponseType> {
     const msgService = new DeleteStateRequest();
     msgService.setStoreName(storeName);
     msgService.setKey(key);
 
-    // @todo: implement below
-    // msgService.setEtag();
-    // msgService.setOptions();
+    if (options?.etag) {
+      const etag = new Etag();
+      etag.setValue(options.etag);
+      msgService.setEtag(etag);
+    }
+
+    msgService.setOptions(this._configureStateOptions(options));
 
     const client = await this.client.getClient();
 
@@ -176,7 +193,7 @@ export default class GRPCClientState implements IClientState {
         }
 
         // https://docs.dapr.io/reference/api/state_api/#http-response-3
-        return resolve();
+        return resolve({});
       });
     });
   }
@@ -200,13 +217,7 @@ export default class GRPCClientState implements IClientState {
         si.setEtag(etag);
       }
 
-      if (o.request.options) {
-        const so = new StateOptions();
-        so.setConsistency(o.request.options.consistency as any);
-        so.setConcurrency(o.request.options.concurrency as any);
-
-        si.setOptions(so);
-      }
+      si.setOptions(this._configureStateOptions(o.request?.options));
 
       const transactionItem = new TransactionalStateOperation();
       transactionItem.setOperationtype(o.operation);
@@ -272,4 +283,21 @@ export default class GRPCClientState implements IClientState {
       });
     });
   }
+
+  _configureStateOptions(opt?: Partial<IStateOptions>): StateOptions | undefined {
+    if (opt === undefined) {
+      return undefined;
+    }
+
+    const stateOptions = new StateOptions();
+    if (opt?.consistency) {
+      stateOptions.setConsistency(opt.consistency as any);
+    }
+
+    if (opt?.concurrency) {
+      stateOptions.setConcurrency(opt.concurrency as any);
+    }
+
+    return stateOptions;
+  }
 }

src/implementation/Client/HTTPClient/state.ts

@@ -20,11 +20,14 @@ import { KeyValueType } from "../../../types/KeyValue.type";
 import { StateQueryType } from "../../../types/state/StateQuery.type";
 import { StateQueryResponseType } from "../../../types/state/StateQueryResponse.type";
 import { StateGetBulkOptions } from "../../../types/state/StateGetBulkOptions.type";
-import { createHTTPMetadataQueryParam } from "../../../utils/Client.util";
+import { createHTTPQueryParam, getStateConcurrencyValue, getStateConsistencyValue } from "../../../utils/Client.util";
 import { Settings } from "../../../utils/Settings.util";
 import { Logger } from "../../../logger/Logger";
 import { StateSaveResponseType } from "../../../types/state/StateSaveResponseType";
 import { StateSaveOptions } from "../../../types/state/StateSaveOptions.type";
+import { StateDeleteOptions } from "../../../types/state/StateDeleteOptions.type";
+import { THTTPExecuteParams } from "../../../types/http/THTTPExecuteParams.type";
+import { StateGetOptions } from "../../../types/state/StateGetOptions.type";
 
 // https://docs.dapr.io/reference/api/state_api/
 export default class HTTPClientState implements IClientState {
@@ -41,7 +44,15 @@ export default class HTTPClientState implements IClientState {
     stateObjects: KeyValuePairType[],
     options: StateSaveOptions = {},
   ): Promise<StateSaveResponseType> {
-    const queryParams = createHTTPMetadataQueryParam(options.metadata);
+    const queryParams = createHTTPQueryParam({ data: options?.metadata, type: "metadata" });
+
+    for (const so of stateObjects) {
+      const behavior = {
+        consistency: getStateConsistencyValue(so?.options?.consistency),
+        concurrency: getStateConcurrencyValue(so?.options?.concurrency),
+      };
+      so.options = Object.assign({}, so.options, behavior);
+    }
 
     try {
       await this.client.execute(`/state/${storeName}?${queryParams}`, {
@@ -56,36 +67,72 @@ export default class HTTPClientState implements IClientState {
     return {};
   }
 
-  async get(storeName: string, key: string): Promise<KeyValueType | string> {
-    const result = await this.client.execute(`/state/${storeName}/${key}`);
+  async get(storeName: string, key: string, options?: Partial<StateGetOptions>): Promise<KeyValueType | string> {
+    const behavior = {
+      consistency: getStateConsistencyValue(options?.consistency),
+    };
+
+    const queryParams = createHTTPQueryParam({ data: options?.metadata, type: "metadata" }, { data: behavior });
+
+    const result = await this.client.execute(`/state/${storeName}/${key}?${queryParams}`);
+
     return result as KeyValueType;
   }
 
-  async getBulk(storeName: string, keys: string[], options: StateGetBulkOptions = {}): Promise<KeyValueType[]> {
-    const queryParams = createHTTPMetadataQueryParam(options.metadata);
+  async getBulk(storeName: string, keys: string[], options?: StateGetBulkOptions): Promise<KeyValueType[]> {
+    const queryParams = createHTTPQueryParam({ data: options?.metadata, type: "metadata" });
 
     const result = await this.client.execute(`/state/${storeName}/bulk?${queryParams}`, {
       method: "POST",
       body: {
         keys,
-        parallelism: options.parallelism ?? Settings.getDefaultStateGetBulkParallelism,
+        parallelism: options?.parallelism ?? Settings.getDefaultStateGetBulkParallelism,
       },
     });
 
     return result as KeyValueType[];
   }
 
-  async delete(storeName: string, key: string): Promise<void> {
-    await this.client.execute(`/state/${storeName}/${key}`, {
-      method: "DELETE",
-    });
+  async delete(storeName: string, key: string, options?: Partial<StateDeleteOptions>): Promise<StateSaveResponseType> {
+    const behavior = {
+      concurrency: getStateConcurrencyValue(options?.concurrency),
+      consistency: getStateConsistencyValue(options?.consistency),
+    };
+
+    const queryParams = createHTTPQueryParam({ data: options?.metadata, type: "metadata" }, { data: behavior });
+
+    // Managed headers
+    const headers: THTTPExecuteParams["headers"] = {};
+    if (options?.etag) {
+      headers["If-Match"] = options.etag;
+    }
+
+    try {
+      await this.client.execute(`/state/${storeName}/${key}?${queryParams}`, {
+        method: "DELETE",
+        headers,
+      });
+    } catch (e: any) {
+      this.logger.error(`Error deleting state from store ${storeName}, error: ${e}`);
+      return { error: e };
+    }
+
+    return {};
   }
 
   async transaction(
     storeName: string,
     operations: OperationType[] = [],
     metadata: IRequestMetadata | null = null,
   ): Promise<void> {
+    for (const op of operations) {
+      const behavior = {
+        consistency: getStateConsistencyValue(op?.request?.options?.consistency),
+        concurrency: getStateConcurrencyValue(op?.request.options?.concurrency),
+      };
+      op.request.options = Object.assign({}, op.request.options, behavior);
+    }
+
     await this.client.execute(`/state/${storeName}/transaction`, {
       method: "POST",
       body: {

src/index.ts

@@ -36,6 +36,8 @@ import { PubSubBulkPublishMessage } from "./types/pubsub/PubSubBulkPublishMessag
 import HttpMethod from "./enum/HttpMethod.enum";
 import CommunicationProtocolEnum from "./enum/CommunicationProtocol.enum";
 import DaprPubSubStatusEnum from "./enum/DaprPubSubStatus.enum";
+import StateConcurrencyEnum from "./enum/StateConcurrency.enum";
+import StateConsistencyEnum from "./enum/StateConsistency.enum";
 import { StateGetBulkOptions } from "./types/state/StateGetBulkOptions.type";
 
 export {
@@ -59,6 +61,8 @@ export {
   CommunicationProtocolEnum,
   DaprPubSubStatusEnum,
   PubSubBulkPublishMessage,
+  StateConcurrencyEnum,
+  StateConsistencyEnum,
   PubSubBulkPublishResponse,
   StateGetBulkOptions,
 };

src/interfaces/Client/IClientState.ts

@@ -20,12 +20,14 @@ import { StateQueryResponseType } from "../../types/state/StateQueryResponse.typ
 import { StateGetBulkOptions } from "../../types/state/StateGetBulkOptions.type";
 import { StateSaveResponseType } from "../../types/state/StateSaveResponseType";
 import { StateSaveOptions } from "../../types/state/StateSaveOptions.type";
+import { StateDeleteOptions } from "../../types/state/StateDeleteOptions.type";
+import { StateGetOptions } from "../../types/state/StateGetOptions.type";
 
 export default interface IClientState {
   save(storeName: string, stateObjects: KeyValuePairType[], options?: StateSaveOptions): Promise<StateSaveResponseType>;
-  get(storeName: string, key: string): Promise<KeyValueType | string>;
+  get(storeName: string, key: string, options?: Partial<StateGetOptions>): Promise<KeyValueType | string>;
   getBulk(storeName: string, keys: string[], options?: StateGetBulkOptions): Promise<KeyValueType[]>;
-  delete(storeName: string, key: string): Promise<void>;
+  delete(storeName: string, key: string, options?: Partial<StateDeleteOptions>): Promise<StateSaveResponseType>;
   transaction(storeName: string, operations?: OperationType[], metadata?: IRequestMetadata | null): Promise<void>;
   query(storeName: string, query: StateQueryType): Promise<StateQueryResponseType>;
 }

src/types/KeyValuePair.type.ts

@@ -12,11 +12,12 @@ limitations under the License.
 */
 
 import { KeyValueType } from "./KeyValue.type";
+import { IStateOptions } from "./state/StateOptions.type";
 
 export type KeyValuePairType = {
   key: string;
   value: any;
   etag?: string;
   metadata?: KeyValueType;
-  options?: object;
+  options?: IStateOptions;
 };

src/types/state/StateDeleteOptions.type.ts

@@ -0,0 +1,28 @@
+/*
+Copyright 2023 The Dapr Authors
+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
+    http://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.
+*/
+
+import { IRequestMetadata } from "../RequestMetadata.type";
+import { IStateOptions } from "./StateOptions.type";
+import { KeyValuePairType } from "../KeyValuePair.type";
+
+export type StateDeleteOptions = IStateOptions & {
+  /**
+   * Metadata to be passed to the operation.
+   */
+  metadata: IRequestMetadata;
+
+  /**
+   * Optional Etag for Optimistic Concurrency Control
+   */
+  etag: KeyValuePairType["etag"];
+};

src/types/state/StateGetOptions.type.ts

@@ -0,0 +1,22 @@
+/*
+Copyright 2023 The Dapr Authors
+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
+    http://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.
+*/
+
+import { IRequestMetadata } from "../RequestMetadata.type";
+import { IStateOptions } from "./StateOptions.type";
+
+export type StateGetOptions = Pick<IStateOptions, "consistency"> & {
+  /**
+   * Metadata to be passed to the operation.
+   */
+  metadata: IRequestMetadata;
+};

src/types/state/StateOptions.type.ts

@@ -11,10 +11,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-import { EStateConsistency } from "../../enum/StateConsistency.enum";
-import { EStateConcurrency } from "../../enum/StateConcurrency.enum";
+import { StateConsistencyEnum } from "../../enum/StateConsistency.enum";
+import { StateConcurrencyEnum } from "../../enum/StateConcurrency.enum";
 
 export type IStateOptions = {
-  concurrency: EStateConcurrency;
-  consistency: EStateConsistency;
+  concurrency: StateConcurrencyEnum;
+  consistency: StateConsistencyEnum;
 };