[PM-15061] extract encryptors from generator service (#12068)

* introduce legacy encryptor provider
* port credential generation service to encryptor provider

libs/common/src/tools/rx.ts

@@ -15,8 +15,60 @@ import {
   takeUntil,
   withLatestFrom,
   concatMap,
+  startWith,
+  pairwise,
 } from "rxjs";
 
+/** Returns its input. */
+function identity(value: any): any {
+  return value;
+}
+
+/** Combines its arguments into a plain old javascript object. */
+function expectedAndActualValue(expectedValue: any, actualValue: any) {
+  return {
+    expectedValue,
+    actualValue,
+  };
+}
+
+/**
+ * An observable operator that throws an error when the stream's
+ *   value changes. Uses strict (`===`) comparison checks.
+ * @param extract a function that identifies the member to compare;
+ *   defaults to the identity function
+ * @param error a function that packages the expected and failed
+ *   values into an error.
+ * @returns a stream of values that emits when the input emits,
+ *   completes when the input completes, and errors when either the
+ *   input errors or the comparison fails.
+ */
+export function errorOnChange<Input, Extracted>(
+  extract: (value: Input) => Extracted = identity,
+  error: (expectedValue: Extracted, actualValue: Extracted) => unknown = expectedAndActualValue,
+): OperatorFunction<Input, Input> {
+  return pipe(
+    startWith(null),
+    pairwise(),
+    map(([expected, actual], i) => {
+      // always let the first value through
+      if (i === 0) {
+        return actual;
+      }
+
+      const expectedValue = extract(expected);
+      const actualValue = extract(actual);
+
+      // fail the stream if the state desyncs from its initial value
+      if (expectedValue === actualValue) {
+        return actual;
+      } else {
+        throw error(expectedValue, actualValue);
+      }
+    }),
+  );
+}
+
 /**
  * An observable operator that reduces an emitted collection to a single object,
  * returning a default if all items are ignored.