Add Automerge CRDT field APIs to package runtime

src/index.ts

@@ -8,6 +8,12 @@ import {
   type HandlerContext,
   type Interceptor,
 } from "@connectrpc/connect";
+import * as Automerge from "@automerge/automerge";
+import {
+  encodeAutomergeFieldWrite,
+  loadAutomergeFieldDoc,
+  type CaminoAutomergeDoc,
+} from "@quixos/camino-datatypes/crdt-automerge";
 import {
   connectNodeAdapter,
   createConnectTransport,
@@ -183,6 +189,11 @@ export type Field<T> = {
   set(value: T): Promise<void>;
 };
 
+export type CrdtField<T> = Field<CaminoAutomergeDoc<T>> & {
+  change(fn: (doc: { value: T }) => void): Promise<CaminoAutomergeDoc<T>>;
+  replaceFromJson(value: T): Promise<void>;
+};
+
 export type LiveFieldValue<T> = {
   $caminoValue: T;
   $caminoSource: {
@@ -475,6 +486,11 @@ const bytesToBase64 = (value: Uint8Array) =>
 
 const base64ToBytes = (value: string) => Buffer.from(value, "base64");
 
+type FieldCodec = {
+  fieldType?: string;
+  conflictStrategy?: string;
+};
+
 const sourceToProto = (value: unknown) => {
   if (!isRecord(value) || !isRecord(value.field)) {
     return undefined;
@@ -507,12 +523,30 @@ const sourceToProto = (value: unknown) => {
   });
 };
 
+const codecFromSource = (value: unknown): FieldCodec | undefined => {
+  if (!isRecord(value) || !isRecord(value.field)) {
+    return undefined;
+  }
+  const field = value.field;
+  return {
+    ...(typeof field.fieldType === "string" ? { fieldType: field.fieldType } : {}),
+    ...(typeof field.conflictStrategy === "string"
+      ? { conflictStrategy: field.conflictStrategy }
+      : {}),
+  };
+};
+
 const withSource = (value: Value, source: unknown): Value => {
   const sourceValue = sourceToProto(source);
   return sourceValue ? create(ValueSchema, { ...value, source: sourceValue }) : value;
 };
 
-export const jsToProtoValue = (value: unknown): Value => {
+export const jsToProtoValue = (value: unknown, codec?: FieldCodec): Value => {
+  if (codec?.conflictStrategy === "crdt" && codec.fieldType) {
+    return jsToProtoValue(
+      encodeAutomergeFieldWrite(value, codec.fieldType),
+    );
+  }
   if (value === null || value === undefined) {
     return create(ValueSchema, {
       kind: {
@@ -551,14 +585,17 @@ export const jsToProtoValue = (value: unknown): Value => {
       kind: {
         case: "listValue",
         value: create(ListValueSchema, {
-          values: value.map(jsToProtoValue),
+          values: value.map((entry) => jsToProtoValue(entry)),
         }),
       },
     });
   }
   if (isRecord(value)) {
     if (Object.hasOwn(value, "$caminoValue") && isRecord(value.$caminoSource)) {
-      return withSource(jsToProtoValue(value.$caminoValue), value.$caminoSource);
+      return withSource(
+        jsToProtoValue(value.$caminoValue, codecFromSource(value.$caminoSource)),
+        value.$caminoSource,
+      );
     }
     if (typeof value.$caminoRef === "string") {
       return create(ValueSchema, {
@@ -631,11 +668,12 @@ export const protoValueToJs = (value: Value | undefined): unknown => {
     case "refValue":
       return value.kind.value.objectId;
     case "crdtValue":
-      return {
+      return loadAutomergeFieldDoc({
         $caminoCrdtType: value.kind.value.type,
-        $caminoCrdtEncoding: value.kind.value.encoding,
+        $caminoCrdtEncoding:
+          value.kind.value.encoding === "base64" ? "base64" : "base64",
         $caminoCrdtPayload: bytesToBase64(value.kind.value.payload),
-      };
+      }, value.kind.value.type);
   }
 };
 
@@ -761,6 +799,7 @@ export const createField = <T>(
   camino: CaminoClient,
   objectId: string,
   fieldName: string,
+  codec?: FieldCodec,
 ): Field<T> => ({
   async get() {
     recordDerivedDependency({ kind: "field", objectId, fieldName });
@@ -771,11 +810,44 @@ export const createField = <T>(
     await camino.setField({
       objectId,
       fieldName,
-      value: jsToProtoValue(value),
+      value: jsToProtoValue(value, codec),
     });
   },
 });
 
+export const createCrdtField = <T>(
+  camino: CaminoClient,
+  objectId: string,
+  fieldName: string,
+  fieldType: string,
+): CrdtField<T> => {
+  const field = createField<CaminoAutomergeDoc<T>>(camino, objectId, fieldName, {
+    fieldType,
+    conflictStrategy: "crdt",
+  });
+  return {
+    ...field,
+    async change(fn) {
+      const current = await field.get();
+      if (!current) {
+        throw new Error(`Cannot change missing CRDT field ${fieldName}`);
+      }
+      const next = Automerge.change(current, (doc) => {
+        fn(doc as { value: T });
+      }) as CaminoAutomergeDoc<T>;
+      await field.set(next);
+      return next;
+    },
+    async replaceFromJson(value) {
+      await camino.setField({
+        objectId,
+        fieldName,
+        value: jsToProtoValue(value, { fieldType, conflictStrategy: "crdt" }),
+      });
+    },
+  };
+};
+
 const createDependencyTrackingCaminoClient = (
   camino: CaminoClient,
 ): CaminoClient =>