Fix kubeconfig-sync sometimes producing multiple identical entities (#5855)

src/common/cluster-store/get-by-id.injectable.ts

@@ -0,0 +1,21 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import type { ClusterId } from "../cluster-types";
+import type { Cluster } from "../cluster/cluster";
+import clusterStoreInjectable from "./cluster-store.injectable";
+
+export type GetClusterById = (id: ClusterId) => Cluster | undefined;
+
+const getClusterByIdInjectable = getInjectable({
+  id: "get-cluster-by-id",
+  instantiate: (di): GetClusterById => {
+    const store = di.inject(clusterStoreInjectable);
+
+    return (id) => store.getById(id);
+  },
+});
+
+export default getClusterByIdInjectable;

src/common/fs/create-read-file-stream.injectable.ts

@@ -0,0 +1,30 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import type { ReadStream } from "fs";
+import fsInjectable from "./fs.injectable";
+
+export interface CreateReadStreamOptions {
+  mode?: number;
+  end?: number | undefined;
+  flags?: string | undefined;
+  encoding?: BufferEncoding | undefined;
+  autoClose?: boolean | undefined;
+  /**
+   * @default false
+   */
+  emitClose?: boolean | undefined;
+  start?: number | undefined;
+  highWaterMark?: number | undefined;
+}
+
+export type CreateReadFileStream = (filePath: string, options?: CreateReadStreamOptions) => ReadStream;
+
+const createReadFileStreamInjectable = getInjectable({
+  id: "create-read-file-stream",
+  instantiate: (di): CreateReadFileStream => di.inject(fsInjectable).createReadStream,
+});
+
+export default createReadFileStreamInjectable;

src/common/fs/stat/stat.injectable.ts

@@ -3,12 +3,14 @@
  * Licensed under MIT License. See LICENSE in root directory for more information.
  */
 import { getInjectable } from "@ogre-tools/injectable";
+import type { Stats } from "fs";
 import fsInjectable from "../fs.injectable";
 
+export type Stat = (path: string) => Promise<Stats>;
+
 const statInjectable = getInjectable({
   id: "stat",
-
-  instantiate: (di) => di.inject(fsInjectable).stat,
+  instantiate: (di): Stat => di.inject(fsInjectable).stat,
 });
 
 export default statInjectable;

src/common/fs/watch/watch.injectable.ts

@@ -3,15 +3,161 @@
  * Licensed under MIT License. See LICENSE in root directory for more information.
  */
 import { getInjectable } from "@ogre-tools/injectable";
-import type { FSWatcher, WatchOptions } from "chokidar";
 import { watch } from "chokidar";
+import type { Stats } from "fs";
+import type TypedEventEmitter from "typed-emitter";
+import type { SingleOrMany } from "../../utils";
 
-export type Watch = (path: string, options?: WatchOptions) => FSWatcher;
+export interface AlwaysStatWatcherEvents {
+  add: (path: string, stats: Stats) => void;
+  addDir: (path: string, stats: Stats) => void;
+  change: (path: string, stats: Stats) => void;
+}
+
+export interface MaybeStatWatcherEvents {
+  add: (path: string, stats?: Stats) => void;
+  addDir: (path: string, stats?: Stats) => void;
+  change: (path: string, stats?: Stats) => void;
+}
+
+export type WatcherEvents<AlwaysStat extends boolean> = BaseWatcherEvents
+  & (
+    AlwaysStat extends true
+      ? AlwaysStatWatcherEvents
+      : MaybeStatWatcherEvents
+  );
+
+export interface BaseWatcherEvents {
+  error: (error: Error) => void;
+  ready: () => void;
+  unlink: (path: string) => void;
+  unlinkDir: (path: string) => void;
+}
+
+export interface Watcher<AlwaysStat extends boolean> extends TypedEventEmitter<WatcherEvents<AlwaysStat>> {
+  close: () => Promise<void>;
+}
+
+export type WatcherOptions<AlwaysStat extends boolean> = {
+  /**
+   * Indicates whether the process should continue to run as long as files are being watched. If
+   * set to `false` when using `fsevents` to watch, no more events will be emitted after `ready`,
+   * even if the process continues to run.
+   */
+  persistent?: boolean;
+
+  /**
+   * ([anymatch](https://github.com/micromatch/anymatch)-compatible definition) Defines files/paths to
+   * be ignored. The whole relative or absolute path is tested, not just filename. If a function
+   * with two arguments is provided, it gets called twice per path - once with a single argument
+   * (the path), second time with two arguments (the path and the
+   * [`fs.Stats`](https://nodejs.org/api/fs.html#fs_class_fs_stats) object of that path).
+   */
+  ignored?: SingleOrMany<string | RegExp | ((path: string) => boolean)>;
+
+  /**
+   * If set to `false` then `add`/`addDir` events are also emitted for matching paths while
+   * instantiating the watching as chokidar discovers these file paths (before the `ready` event).
+   */
+  ignoreInitial?: boolean;
+
+  /**
+   * When `false`, only the symlinks themselves will be watched for changes instead of following
+   * the link references and bubbling events through the link's path.
+   */
+  followSymlinks?: boolean;
+
+  /**
+   * The base directory from which watch `paths` are to be derived. Paths emitted with events will
+   * be relative to this.
+   */
+  cwd?: string;
+
+  /**
+   *  If set to true then the strings passed to .watch() and .add() are treated as literal path
+   *  names, even if they look like globs. Default: false.
+   */
+  disableGlobbing?: boolean;
+
+  /**
+   * Whether to use fs.watchFile (backed by polling), or fs.watch. If polling leads to high CPU
+   * utilization, consider setting this to `false`. It is typically necessary to **set this to
+   * `true` to successfully watch files over a network**, and it may be necessary to successfully
+   * watch files in other non-standard situations. Setting to `true` explicitly on OS X overrides
+   * the `useFsEvents` default.
+   */
+  usePolling?: boolean;
+
+  /**
+   * Whether to use the `fsevents` watching interface if available. When set to `true` explicitly
+   * and `fsevents` is available this supercedes the `usePolling` setting. When set to `false` on
+   * OS X, `usePolling: true` becomes the default.
+   */
+  useFsEvents?: boolean;
+
+  /**
+   * If set, limits how many levels of subdirectories will be traversed.
+   */
+  depth?: number;
+
+  /**
+   * Interval of file system polling.
+   */
+  interval?: number;
+
+  /**
+   * Interval of file system polling for binary files. ([see list of binary extensions](https://gi
+   * thub.com/sindresorhus/binary-extensions/blob/master/binary-extensions.json))
+   */
+  binaryInterval?: number;
+
+  /**
+   *  Indicates whether to watch files that don't have read permissions if possible. If watching
+   *  fails due to `EPERM` or `EACCES` with this set to `true`, the errors will be suppressed
+   *  silently.
+   */
+  ignorePermissionErrors?: boolean;
+
+  /**
+   * `true` if `useFsEvents` and `usePolling` are `false`). Automatically filters out artifacts
+   * that occur when using editors that use "atomic writes" instead of writing directly to the
+   * source file. If a file is re-added within 100 ms of being deleted, Chokidar emits a `change`
+   * event rather than `unlink` then `add`. If the default of 100 ms does not work well for you,
+   * you can override it by setting `atomic` to a custom value, in milliseconds.
+   */
+  atomic?: boolean | number;
+
+  /**
+   * can be set to an object in order to adjust timing params:
+   */
+  awaitWriteFinish?: AwaitWriteFinishOptions | boolean;
+} & (AlwaysStat extends true
+  ? {
+    alwaysStat: true;
+  }
+  : {
+    alwaysStat?: false;
+  }
+);
+
+export interface AwaitWriteFinishOptions {
+  /**
+   * Amount of time in milliseconds for a file size to remain constant before emitting its event.
+   */
+  stabilityThreshold?: number;
+
+  /**
+   * File size polling interval.
+   */
+  pollInterval?: number;
+}
+
+export type Watch = <AlwaysStat extends boolean = false>(path: string, options?: WatcherOptions<AlwaysStat>) => Watcher<AlwaysStat>;
 
 // TODO: Introduce wrapper to allow simpler API
 const watchInjectable = getInjectable({
   id: "watch",
-  instantiate: (): Watch => watch,
+  instantiate: () => watch as Watch,
   causesSideEffects: true,
 });
 

src/common/logger/prefixed-logger.injectable.ts

@@ -0,0 +1,37 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable, lifecycleEnum } from "@ogre-tools/injectable";
+import type { Logger } from "../logger";
+import loggerInjectable from "../logger.injectable";
+
+const prefixedLoggerInjectable = getInjectable({
+  id: "prefixed-logger",
+  instantiate: (di, prefix): Logger => {
+    const logger = di.inject(loggerInjectable);
+
+    return {
+      debug: (message, ...args) => {
+        logger.debug(`[${prefix}]: ${message}`, ...args);
+      },
+      error: (message, ...args) => {
+        logger.error(`[${prefix}]: ${message}`, ...args);
+      },
+      info: (message, ...args) => {
+        logger.info(`[${prefix}]: ${message}`, ...args);
+      },
+      silly: (message, ...args) => {
+        logger.silly(`[${prefix}]: ${message}`, ...args);
+      },
+      warn: (message, ...args) => {
+        logger.warn(`[${prefix}]: ${message}`, ...args);
+      },
+    };
+  },
+  lifecycle: lifecycleEnum.keyedSingleton({
+    getInstanceKey: (di, prefix: string) => prefix,
+  }),
+});
+
+export default prefixedLoggerInjectable;

src/common/runnable/run-many-for.test.ts

@@ -25,13 +25,19 @@ describe("runManyFor", () => {
 
       const someInjectable = getInjectable({
         id: "some-injectable",
-        instantiate: () => ({ run: () => runMock("some-call") }),
+        instantiate: () => ({
+          id: "some-injectable",
+          run: () => runMock("some-call"),
+        }),
         injectionToken: someInjectionTokenForRunnables,
       });
 
       const someOtherInjectable = getInjectable({
         id: "some-other-injectable",
-        instantiate: () => ({ run: () => runMock("some-other-call") }),
+        instantiate: () => ({
+          id: "some-other-injectable",
+          run: () => runMock("some-other-call"),
+        }),
         injectionToken: someInjectionTokenForRunnables,
       });
 
@@ -79,6 +85,7 @@ describe("runManyFor", () => {
         id: "some-injectable-1",
 
         instantiate: (di) => ({
+          id: "some-injectable-1",
           run: () => runMock("third-level-run"),
           runAfter: di.inject(someInjectable2),
         }),
@@ -90,6 +97,7 @@ describe("runManyFor", () => {
         id: "some-injectable-2",
 
         instantiate: (di) => ({
+          id: "some-injectable-2",
           run: () => runMock("second-level-run"),
           runAfter: di.inject(someInjectable3),
         }),
@@ -99,7 +107,10 @@ describe("runManyFor", () => {
 
       const someInjectable3 = getInjectable({
         id: "some-injectable-3",
-        instantiate: () => ({ run: () => runMock("first-level-run") }),
+        instantiate: () => ({
+          id: "some-injectable-3",
+          run: () => runMock("first-level-run"),
+        }),
         injectionToken: someInjectionTokenForRunnables,
       });
 
@@ -186,6 +197,7 @@ describe("runManyFor", () => {
       id: "some-runnable-1",
 
       instantiate: (di) => ({
+        id: "some-runnable-1",
         run: () => runMock("some-runnable-1"),
         runAfter: di.inject(someOtherInjectable),
       }),
@@ -197,6 +209,7 @@ describe("runManyFor", () => {
       id: "some-runnable-2",
 
       instantiate: () => ({
+        id: "some-runnable-2",
         run: () => runMock("some-runnable-2"),
       }),
 
@@ -210,7 +223,7 @@ describe("runManyFor", () => {
     );
 
     return expect(() => runMany()).rejects.toThrow(
-      "Tried to run runnable after other runnable which does not same injection token.",
+      'Tried to run runnable "some-runnable-1" after the runnable "some-runnable-2" which does not share the "some-injection-token" injection token.',
     );
   });
 
@@ -232,6 +245,7 @@ describe("runManyFor", () => {
         id: "some-runnable-1",
 
         instantiate: () => ({
+          id: "some-runnable-1",
           run: (parameter) => runMock("run-of-some-runnable-1", parameter),
         }),
 
@@ -242,6 +256,7 @@ describe("runManyFor", () => {
         id: "some-runnable-2",
 
         instantiate: () => ({
+          id: "some-runnable-2",
           run: (parameter) => runMock("run-of-some-runnable-2", parameter),
         }),
 

src/common/runnable/run-many-for.ts

@@ -11,8 +11,9 @@ import { filter, forEach, map, tap } from "lodash/fp";
 import { throwWithIncorrectHierarchyFor } from "./throw-with-incorrect-hierarchy-for";
 
 export interface Runnable<TParameter = void> {
+  id: string;
   run: Run<TParameter>;
-  runAfter?: this;
+  runAfter?: Runnable<TParameter>;
 }
 
 type Run<Param> = (parameter: Param) => Promise<void> | void;
@@ -25,7 +26,7 @@ export function runManyFor(di: DiContainerForInjection): RunMany {
   return (injectionToken) => async (parameter) => {
     const allRunnables = di.injectMany(injectionToken);
 
-    const throwWithIncorrectHierarchy = throwWithIncorrectHierarchyFor(allRunnables);
+    const throwWithIncorrectHierarchy = throwWithIncorrectHierarchyFor((injectionToken as any).id, allRunnables);
 
     const recursedRun = async (
       runAfterRunnable: Runnable<any> | undefined = undefined,

src/common/runnable/run-many-sync-for.test.ts

@@ -21,13 +21,19 @@ describe("runManySyncFor", () => {
 
       const someInjectable = getInjectable({
         id: "some-injectable",
-        instantiate: () => ({ run: () => runMock("some-call") }),
+        instantiate: () => ({
+          id: "some-injectable",
+          run: () => runMock("some-call"),
+        }),
         injectionToken: someInjectionTokenForRunnables,
       });
 
       const someOtherInjectable = getInjectable({
         id: "some-other-injectable",
-        instantiate: () => ({ run: () => runMock("some-other-call") }),
+        instantiate: () => ({
+          id: "some-other-injectable",
+          run: () => runMock("some-other-call"),
+        }),
         injectionToken: someInjectionTokenForRunnables,
       });
 
@@ -62,6 +68,7 @@ describe("runManySyncFor", () => {
         id: "some-injectable-1",
 
         instantiate: (di) => ({
+          id: "some-injectable-1",
           run: () => runMock("third-level-run"),
           runAfter: di.inject(someInjectable2),
         }),
@@ -73,6 +80,7 @@ describe("runManySyncFor", () => {
         id: "some-injectable-2",
 
         instantiate: (di) => ({
+          id: "some-injectable-2",
           run: () => runMock("second-level-run"),
           runAfter: di.inject(someInjectable3),
         }),
@@ -82,7 +90,10 @@ describe("runManySyncFor", () => {
 
       const someInjectable3 = getInjectable({
         id: "some-injectable-3",
-        instantiate: () => ({ run: () => runMock("first-level-run") }),
+        instantiate: () => ({
+          id: "some-injectable-3",
+          run: () => runMock("first-level-run"),
+        }),
         injectionToken: someInjectionTokenForRunnables,
       });
 
@@ -115,6 +126,7 @@ describe("runManySyncFor", () => {
       id: "some-runnable-1",
 
       instantiate: (di) => ({
+        id: "some-runnable-1",
         run: () => runMock("some-runnable-1"),
         runAfter: di.inject(someOtherInjectable),
       }),
@@ -126,6 +138,7 @@ describe("runManySyncFor", () => {
       id: "some-runnable-2",
 
       instantiate: () => ({
+        id: "some-runnable-2",
         run: () => runMock("some-runnable-2"),
       }),
 
@@ -139,7 +152,7 @@ describe("runManySyncFor", () => {
     );
 
     return expect(() => runMany()).rejects.toThrow(
-      "Tried to run runnable after other runnable which does not same injection token.",
+      'Tried to run runnable "some-runnable-1" after the runnable "some-runnable-2" which does not share the "some-injection-token" injection token.',
     );
   });
 
@@ -161,6 +174,7 @@ describe("runManySyncFor", () => {
         id: "some-runnable-1",
 
         instantiate: () => ({
+          id: "some-runnable-1",
           run: (parameter) => runMock("run-of-some-runnable-1", parameter),
         }),
 
@@ -171,6 +185,7 @@ describe("runManySyncFor", () => {
         id: "some-runnable-2",
 
         instantiate: () => ({
+          id: "some-runnable-2",
           run: (parameter) => runMock("run-of-some-runnable-2", parameter),
         }),
 

src/common/runnable/run-many-sync-for.ts

@@ -3,17 +3,15 @@
  * Licensed under MIT License. See LICENSE in root directory for more information.
  */
 import { pipeline } from "@ogre-tools/fp";
-import type {
-  DiContainerForInjection,
-  InjectionToken,
-} from "@ogre-tools/injectable";
+import type { DiContainerForInjection, InjectionToken } from "@ogre-tools/injectable";
 import { filter, forEach, map, tap } from "lodash/fp";
 import type { Runnable } from "./run-many-for";
 import { throwWithIncorrectHierarchyFor } from "./throw-with-incorrect-hierarchy-for";
 
 export interface RunnableSync<TParameter = void> {
+  id: string;
   run: RunSync<TParameter>;
-  runAfter?: this;
+  runAfter?: RunnableSync<TParameter>;
 }
 
 type RunSync<Param> = (parameter: Param) => void;
@@ -26,7 +24,7 @@ export function runManySyncFor(di: DiContainerForInjection): RunManySync {
   return (injectionToken) => async (parameter) => {
     const allRunnables = di.injectMany(injectionToken);
 
-    const throwWithIncorrectHierarchy = throwWithIncorrectHierarchyFor(allRunnables);
+    const throwWithIncorrectHierarchy = throwWithIncorrectHierarchyFor((injectionToken as any).id, allRunnables);
 
     const recursedRun = (
       runAfterRunnable: RunnableSync<any> | undefined = undefined,

src/common/runnable/throw-with-incorrect-hierarchy-for.ts

@@ -5,12 +5,10 @@
 import type { Runnable } from "./run-many-for";
 import type { RunnableSync } from "./run-many-sync-for";
 
-export const throwWithIncorrectHierarchyFor =
-  (allRunnables: Runnable<any>[] | RunnableSync<any>[]) =>
-    (runnable: Runnable<any> | RunnableSync<any>) => {
-      if (runnable.runAfter && !allRunnables.includes(runnable.runAfter)) {
-        throw new Error(
-          "Tried to run runnable after other runnable which does not same injection token.",
-        );
-      }
-    };
+export const throwWithIncorrectHierarchyFor = (injectionTokenId: string, allRunnables: Runnable<any>[] | RunnableSync<any>[]) => (
+  (runnable: Runnable<any> | RunnableSync<any>) => {
+    if (runnable.runAfter && !allRunnables.includes(runnable.runAfter)) {
+      throw new Error(`Tried to run runnable "${runnable.id}" after the runnable "${runnable.runAfter.id}" which does not share the "${injectionTokenId}" injection token.`);
+    }
+  }
+);

src/common/user-store/kubeconfig-syncs.injectable.ts

@@ -0,0 +1,17 @@
+/**
+ * Copyright (c) OpenLens Authors. All rights reserved.
+ * Licensed under MIT License. See LICENSE in root directory for more information.
+ */
+import { getInjectable } from "@ogre-tools/injectable";
+import userStoreInjectable from "./user-store.injectable";
+
+const kubeconfigSyncsInjectable = getInjectable({
+  id: "kubeconfig-syncs",
+  instantiate: (di) => {
+    const store = di.inject(userStoreInjectable);
+
+    return store.syncKubeconfigEntries;
+  },
+});
+
+export default kubeconfigSyncsInjectable;

src/common/utils/iter.ts

@@ -11,6 +11,7 @@ interface Iterator<T> {
   find(fn: (val: T) => unknown): T | undefined;
   collect<U>(fn: (values: Iterable<T>) => U): U;
   map<U>(fn: (val: T) => U): Iterator<U>;
+  flatMap<U>(fn: (val: T) => U[]): Iterator<U>;
   join(sep?: string): string;
 }
 
@@ -19,6 +20,7 @@ export function pipeline<T>(src: IterableIterator<T>): Iterator<T> {
     filter: (fn) => pipeline(filter(src, fn)),
     filterMap: (fn) => pipeline(filterMap(src, fn)),
     map: (fn) => pipeline(map(src, fn)),
+    flatMap: (fn) => pipeline(flatMap(src, fn)),
     find: (fn) => find(src, fn),
     join: (sep) => join(src, sep),
     collect: (fn) => fn(src),