rewrite2

src/custom-resouces/authentik-client/authentik-client.controller.ts

@@ -17,7 +17,7 @@ import { authentikServerSecretSchema } from '../authentik-server/authentik-serve
 
 import { authentikClientSecretSchema, type authentikClientSpecSchema } from './authentik-client.schemas.ts';
 
-class AuthentikClientResource extends CustomResource<typeof authentikClientSpecSchema> {
+class AuthentikClientController extends CustomResource<typeof authentikClientSpecSchema> {
   #serverSecret: ResourceReference<V1Secret>;
   #clientSecretResource: Resource<V1Secret>;
 
@@ -172,4 +172,4 @@ class AuthentikClientResource extends CustomResource<typeof authentikClientSpecS
   };
 }
 
-export { AuthentikClientResource };
+export { AuthentikClientController };

src/custom-resouces/authentik-client/authentik-client.ts

@@ -1,7 +1,7 @@
 import { createCustomResourceDefinition } from '../../services/custom-resources/custom-resources.ts';
 import { GROUP } from '../../utils/consts.ts';
 
-import { AuthentikClientResource } from './authentik-client.resource.ts';
+import { AuthentikClientController } from './authentik-client.controller.ts';
 import { authentikClientSpecSchema } from './authentik-client.schemas.ts';
 
 const authentikClientDefinition = createCustomResourceDefinition({
@@ -12,7 +12,7 @@ const authentikClientDefinition = createCustomResourceDefinition({
     plural: 'authentikclients',
     singular: 'authentikclient',
   },
-  create: (options) => new AuthentikClientResource(options),
+  create: (options) => new AuthentikClientController(options),
   spec: authentikClientSpecSchema,
 });
 

src/custom-resouces/authentik-server/authentik-server.controller.ts

@@ -93,10 +93,11 @@ class AuthentikServerController extends CustomResource<typeof authentikServerSpe
     this.#redisServer = new ResourceReference();
     this.#postgresSecret = new ResourceReference();
     this.#authentikSecret.on('changed', this.queueReconcile);
-    this.#authentikInitSecret.resource.on('deleted', this.queueReconcile);
+    this.#authentikInitSecret.resource.on('changed', this.queueReconcile);
     this.#environment.on('changed', this.queueReconcile);
     this.#authentikRelease.on('changed', this.queueReconcile);
     this.#postgresSecret.on('changed', this.queueReconcile);
+    this.#postgresDatabase.on('changed', this.queueReconcile);
     this.#httpService.on('changed', this.queueReconcile);
     this.#redisServer.on('changed', this.queueReconcile);
   }

src/instances/authentik-server.ts

@@ -1,7 +1,26 @@
-import type { authentikServerSpecSchema } from '../custom-resouces/authentik-server/authentik-server.schemas.ts';
+import {
+  authentikServerSecretSchema,
+  type authentikServerSpecSchema,
+} from '../custom-resouces/authentik-server/authentik-server.schemas.ts';
 import type { CustomResourceObject } from '../services/custom-resources/custom-resources.custom-resource.ts';
 import { ResourceInstance } from '../services/resources/resources.instance.ts';
+import { ResourceService } from '../services/resources/resources.ts';
 
-class AuthentikServerInstance extends ResourceInstance<CustomResourceObject<typeof authentikServerSpecSchema>> {}
+import { SecretInstance } from './secret.ts';
+
+class AuthentikServerInstance extends ResourceInstance<CustomResourceObject<typeof authentikServerSpecSchema>> {
+  public get secret() {
+    const resourceService = this.services.get(ResourceService);
+    return resourceService.getInstance(
+      {
+        apiVersion: 'v1',
+        kind: 'Secret',
+        name: `${this.name}-server`,
+        namespace: this.namespace,
+      },
+      SecretInstance<typeof authentikServerSecretSchema>,
+    );
+  }
+}
 
 export { AuthentikServerInstance };

src/instances/postgres-database.ts

@@ -1,3 +1,4 @@
+import { postgresClusterSecretSchema } from '../custom-resouces/postgres-cluster/postgres-cluster.schemas.ts';
 import type { postgresDatabaseSpecSchema } from '../custom-resouces/postgres-database/portgres-database.schemas.ts';
 import type { CustomResourceObject } from '../services/custom-resources/custom-resources.custom-resource.ts';
 import { ResourceInstance } from '../services/resources/resources.instance.ts';
@@ -15,7 +16,7 @@ class PostgresDatabaseInstance extends ResourceInstance<CustomResourceObject<typ
         name: `${this.name}-postgres-database`,
         namespace: this.namespace,
       },
-      SecretInstance,
+      SecretInstance<typeof postgresClusterSecretSchema>,
     );
   }
 }

src/services/controllers/controllers.dependencies.ts

@@ -0,0 +1,15 @@
+import type { ResourceInstance } from '../resources/resources.instance.ts';
+
+type DependencyRef<T extends ResourceInstance<ExpectedAny>> = {
+  apiVersion: string;
+  kind: string;
+  name: string;
+  namespace?: string;
+  instance: T;
+};
+
+class CustomResourceControllerDependencies {
+  public get = <T extends ResourceInstance<ExpectedAny>>(name: string, ref: DependencyRef<T>) => { };
+}
+
+export { CustomResourceControllerDependencies };

src/services/controllers/controllers.types.ts

@@ -0,0 +1,25 @@
+import type { z, ZodAny, ZodType } from 'zod';
+import type { KubernetesObject } from '@kubernetes/client-node';
+
+import type { Resource } from '../resources/resources.resource.ts';
+
+import type { CustomResourceControllerDependencies } from './controllers.dependencies.ts';
+
+type CustomResourceControllerOptions<TSpec extends ZodType> = {
+  resource: Resource<KubernetesObject & { spec: z.infer<TSpec> }>;
+  dependencies: CustomResourceControllerDependencies;
+};
+
+type CustomResourceController<TSpec extends ZodType> = (options: CustomResourceControllerOptions<TSpec>) => {
+  reconcile: () => Promise<void>;
+};
+
+type CustomResource<TSpec extends ZodAny> = {
+  group: string;
+  version: string;
+  spec: TSpec;
+  scope: 'namespace' | 'cluster';
+  controller: CustomResourceController<TSpec>;
+};
+
+export type { CustomResource, CustomResourceController };

src/services/resources/resource/resource.ts

@@ -0,0 +1,172 @@
+import { ApiException, PatchStrategy, type KubernetesObject } from '@kubernetes/client-node';
+import { EventEmitter } from 'eventemitter3';
+import deepEqual from 'deep-equal';
+
+import type { Services } from '../../../utils/service.ts';
+import { Queue } from '../../queue/queue.ts';
+import { K8sService } from '../../k8s/k8s.ts';
+import { isDeepSubset } from '../../../utils/objects.ts';
+
+type ResourceSelector = {
+  apiVersion: string;
+  kind: string;
+  name: string;
+  namespace?: string;
+};
+
+type ResourceOptions<T extends KubernetesObject> = {
+  services: Services;
+  selector: ResourceSelector;
+  manifest?: T;
+};
+
+type ResourceEvents = {
+  changed: () => void;
+};
+
+class Resource<T extends KubernetesObject> extends EventEmitter<ResourceEvents> {
+  #manifest?: T;
+  #queue: Queue;
+  #options: ResourceOptions<T>;
+
+  constructor(options: ResourceOptions<T>) {
+    super();
+    this.#options = options;
+    this.#manifest = options.manifest;
+    this.#queue = new Queue({ concurrency: 1 });
+  }
+
+  public get manifest() {
+    return this.#manifest;
+  }
+
+  public set manifest(value: T | undefined) {
+    if (deepEqual(this.manifest, value)) {
+      return;
+    }
+    this.#manifest = value;
+    this.emit('changed');
+  }
+
+  public get exists() {
+    return !!this.#manifest;
+  }
+
+  public get ready() {
+    return this.exists;
+  }
+
+  public get selector() {
+    return this.#options.selector;
+  }
+
+  public get apiVersion() {
+    return this.selector.apiVersion;
+  }
+
+  public get kind() {
+    return this.selector.kind;
+  }
+
+  public get name() {
+    return this.selector.name;
+  }
+
+  public get namespace() {
+    return this.selector.namespace;
+  }
+
+  public get metadata() {
+    return this.manifest?.metadata;
+  }
+
+  public get ref() {
+    if (!this.metadata?.uid) {
+      throw new Error('No uid for resource');
+    }
+    return {
+      apiVersion: this.apiVersion,
+      kind: this.kind,
+      name: this.name,
+      uid: this.metadata.uid,
+    };
+  }
+
+  public get spec(): (T extends { spec?: infer K } ? K : never) | undefined {
+    const manifest = this.manifest;
+    if (!manifest || !('spec' in manifest)) {
+      return;
+    }
+    return manifest.spec as ExpectedAny;
+  }
+
+  public get data(): (T extends { data?: infer K } ? K : never) | undefined {
+    const manifest = this.manifest;
+    if (!manifest || !('data' in manifest)) {
+      return;
+    }
+    return manifest.data as ExpectedAny;
+  }
+
+  public get status(): (T extends { status?: infer K } ? K : never) | undefined {
+    const manifest = this.manifest;
+    if (!manifest || !('status' in manifest)) {
+      return;
+    }
+    return manifest.status as ExpectedAny;
+  }
+
+  public patch = (patch: T) =>
+    this.#queue.add(async () => {
+      const { services } = this.#options;
+      services.log.debug(`Patching ${this.apiVersion}/${this.kind}/${this.namespace}/${this.name}`, {
+        spelector: this.selector,
+        current: this.manifest,
+        patch,
+      });
+      const k8s = services.get(K8sService);
+      const body = {
+        ...patch,
+        apiVersion: this.selector.apiVersion,
+        kind: this.selector.kind,
+        metadata: {
+          ...patch.metadata,
+          name: this.selector.name,
+          namespace: this.selector.namespace,
+        },
+      };
+      try {
+        this.manifest = await k8s.objectsApi.patch(
+          body,
+          undefined,
+          undefined,
+          undefined,
+          undefined,
+          PatchStrategy.MergePatch,
+        );
+      } catch (err) {
+        if (err instanceof ApiException && err.code === 404) {
+          this.manifest = await k8s.objectsApi.create(body);
+          return;
+        }
+        throw err;
+      }
+    });
+
+  public getCondition = (
+    condition: string,
+  ): T extends { status?: { conditions?: (infer U)[] } } ? U | undefined : undefined => {
+    const status = this.status as ExpectedAny;
+    return status?.conditions?.find((c: ExpectedAny) => c?.type === condition);
+  };
+
+  public ensure = async (manifest: T) => {
+    if (isDeepSubset(this.manifest, manifest)) {
+      return false;
+    }
+    await this.patch(manifest);
+    return true;
+  };
+}
+
+export { Resource, type ResourceOptions };

src/services/resources/resources.ts

@@ -1,54 +1,43 @@
 import type { KubernetesObject } from '@kubernetes/client-node';
 
 import type { Services } from '../../utils/service.ts';
+import { WatcherService } from '../watchers/watchers.ts';
 
-import { Resource } from './resources.resource.ts';
-import type { ResourceInstance } from './resources.instance.ts';
+import type { Resource, ResourceOptions } from './resource/resource.ts';
 
-type ResourceGetOptions = {
+type ResourceClass<T extends KubernetesObject> = new (options: ResourceOptions<T>) => Resource<T>;
+
+type RegisterOptions<T extends KubernetesObject> = {
   apiVersion: string;
   kind: string;
-  name: string;
-  namespace?: string;
+  plural?: string;
+  type: ResourceClass<T>;
 };
 
 class ResourceService {
-  #cache: Resource<ExpectedAny>[] = [];
   #services: Services;
+  #registry: Map<Resource<ExpectedAny>, Resource<ExpectedAny>[]>;
 
   constructor(services: Services) {
     this.#services = services;
+    this.#registry = new Map();
   }
 
-  public getInstance = <T extends KubernetesObject, I extends ResourceInstance<T>>(
-    options: ResourceGetOptions,
-    instance: new (resource: Resource<T>) => I,
-  ) => {
-    const resource = this.get<T>(options);
-    return new instance(resource);
+  public register = async <T extends KubernetesObject>(options: RegisterOptions<T>) => {
+    const watcherService = this.#services.get(WatcherService);
+    const watcher = watcherService.create({});
+    watcher.on('changed', (manifest) => {
+      const { name, namespace } = manifest.metadata || {};
+      if (!name) {
+        return;
+      }
+      const current = this.get(options.type, name, namespace);
+      current.manifest = manifest;
+    });
+    await watcher.start();
   };
 
-  public get = <T extends KubernetesObject>(options: ResourceGetOptions) => {
-    const { apiVersion, kind, name, namespace } = options;
-    let resource = this.#cache.find(
-      (resource) =>
-        resource.specifier.kind === kind &&
-        resource.specifier.apiVersion === apiVersion &&
-        resource.specifier.name === name &&
-        resource.specifier.namespace === namespace,
-    );
-    if (resource) {
-      return resource as Resource<T>;
-    }
-    resource = new Resource({
-      data: options,
-      services: this.#services,
-    });
-    this.#cache.push(resource);
-    return resource as Resource<T>;
-  };
+  public get = <T extends KubernetesObject>(type: ResourceClass<T>, name: string, namespace?: string) => {};
 }
 
-export { ResourceInstance } from './resources.instance.ts';
-export { ResourceReference } from './resources.ref.ts';
-export { ResourceService, Resource };
+export { ResourceService };

src/services/resources_old/resources.ts

@@ -0,0 +1,54 @@
+import type { KubernetesObject } from '@kubernetes/client-node';
+
+import type { Services } from '../../utils/service.ts';
+
+import { Resource } from './resources.resource.ts';
+import type { ResourceInstance } from './resources.instance.ts';
+
+type ResourceGetOptions = {
+  apiVersion: string;
+  kind: string;
+  name: string;
+  namespace?: string;
+};
+
+class ResourceService {
+  #cache: Resource<ExpectedAny>[] = [];
+  #services: Services;
+
+  constructor(services: Services) {
+    this.#services = services;
+  }
+
+  public getInstance = <T extends KubernetesObject, I extends ResourceInstance<T>>(
+    options: ResourceGetOptions,
+    instance: new (resource: Resource<T>) => I,
+  ) => {
+    const resource = this.get<T>(options);
+    return new instance(resource);
+  };
+
+  public get = <T extends KubernetesObject>(options: ResourceGetOptions) => {
+    const { apiVersion, kind, name, namespace } = options;
+    let resource = this.#cache.find(
+      (resource) =>
+        resource.specifier.kind === kind &&
+        resource.specifier.apiVersion === apiVersion &&
+        resource.specifier.name === name &&
+        resource.specifier.namespace === namespace,
+    );
+    if (resource) {
+      return resource as Resource<T>;
+    }
+    resource = new Resource({
+      data: options,
+      services: this.#services,
+    });
+    this.#cache.push(resource);
+    return resource as Resource<T>;
+  };
+}
+
+export { ResourceInstance } from './resources.instance.ts';
+export { ResourceReference } from './resources.ref.ts';
+export { ResourceService, Resource };

src/services/watchers/watchers.watcher.ts

@@ -9,12 +9,11 @@ import { EventEmitter } from 'eventemitter3';
 
 import { K8sService } from '../k8s/k8s.ts';
 import type { Services } from '../../utils/service.ts';
-import { ResourceService, type Resource } from '../resources/resources.ts';
 
 type ResourceChangedAction = 'add' | 'update' | 'delete';
 
 type WatcherEvents<T extends KubernetesObject> = {
-  changed: (resource: Resource<T>) => void;
+  changed: (manifest: T) => void;
 };
 
 type WatcherOptions<T extends KubernetesObject = KubernetesObject> = {
@@ -53,27 +52,10 @@ class Watcher<T extends KubernetesObject> extends EventEmitter<WatcherEvents<T>>
   };
 
   #handleResource = (action: ResourceChangedAction, originalManifest: T) => {
-    const { services, transform } = this.#options;
+    const { transform } = this.#options;
     const manifest = transform ? transform(originalManifest) : originalManifest;
-    const resourceService = services.get(ResourceService);
-    const { apiVersion, kind, metadata = {} } = manifest;
-    const { name, namespace } = metadata;
-    if (!name || !apiVersion || !kind) {
-      return;
-    }
-    const resource = resourceService.get<T>({
-      apiVersion,
-      kind,
-      name,
-      namespace,
-    });
 
-    if (action === 'delete') {
-      resource.manifest = undefined;
-    } else {
-      resource.manifest = manifest;
-    }
-    this.emit('changed', resource);
+    this.emit('changed', manifest);
   };
 
   public stop = async () => {