feat: File sync

src/BackgroundSync.ts

@@ -1,14 +1,19 @@
-import { requestUrl, type RequestUrlResponse } from "obsidian";
+import {
+	arrayBufferToBase64,
+	requestUrl,
+	type RequestUrlResponse,
+} from "obsidian";
 import type { HasProvider } from "./HasProvider";
 import type { LoginManager } from "./LoginManager";
 import * as Y from "yjs";
 import { S3RN, S3RemoteDocument, S3RemoteFolder } from "./S3RN";
 import type { SharedFolder, SharedFolders } from "./SharedFolder";
-import type { Document } from "./Document";
+import { Document } from "./Document";
 import type { TimeProvider } from "./TimeProvider";
 import { RelayInstances, curryLog } from "./debug";
 import type { Unsubscriber } from "./observable/Observable";
 import { diff_match_patch, type Diff } from "diff-match-patch";
+import { SyncFile } from "./SyncFile";
 
 declare const API_URL: string;
 
@@ -76,6 +81,7 @@ export class BackgroundSync {
 	log = curryLog("[BackgroundSync]", "log");
 	debug = curryLog("[BackgroundSync]", "debug");
 	error = curryLog("[BackgroundSync]", "error");
+
 	constructor(
 		private loginManager: LoginManager,
 		private timeProvider: TimeProvider,
@@ -175,7 +181,9 @@ export class BackgroundSync {
 			const newDoc = new Y.Doc();
 			Y.applyUpdate(newDoc, updateBytes);
 			if (!newDoc.getText("contents").toString() && hasContents) {
-				this.log("[getDocument] server contents empty document, not overwriting local file.");
+				this.log(
+					"[getDocument] server contents empty document, not overwriting local file.",
+				);
 				return;
 			}
 
@@ -186,7 +194,7 @@ export class BackgroundSync {
 				this.log("Skipping flush - file requires merge conflict resolution.");
 				return;
 			}
-            doc.sharedFolder.flush(doc, doc.text);
+			doc.sharedFolder.flush(doc, doc.text);
 		} catch (e) {
 			console.error(e);
 			return;
@@ -211,13 +219,15 @@ export class BackgroundSync {
 		await folder.whenReady();
 		this.log("[putFolderFiles]", `Uploading ${folder.docset.size} items`);
 		let i = 1;
-		for (const doc of folder.docset.items()) {
-			await doc.whenReady();
-			if (doc.text) {
-				await this.uploadItem(doc);
+		for (const file of folder.docset.items()) {
+			if (file instanceof Document) {
+				await file.whenReady();
+				if (file.text) {
+					await this.uploadItem(file);
+				}
+				this.log("[putFolderFiles]", `${i}/${folder.docset.size}`);
+				i++;
 			}
-			this.log("[putFolderFiles]", `${i}/${folder.docset.size}`);
-			i++;
 		}
 	}
 
@@ -230,12 +240,20 @@ export class BackgroundSync {
 
 	async getFolderFiles(folder: SharedFolder) {
 		await folder.whenReady();
+		if (!folder.shouldConnect) {
+			return;
+		}
 		this.log("[getFolderFiles]", `Downloading ${folder.docset.size} files`);
 		let i = 1;
-		for (const doc of folder.docset.items()) {
-			await this.getDocument(doc);
-			this.log("[getFolderFiles]", `${i}/${folder.docset.size}`);
-			i++;
+		for (const file of folder.docset.items()) {
+			if (file instanceof Document) {
+				await this.getDocument(file);
+				this.log("[getFolderFiles]", `${i}/${folder.docset.size}`);
+				i++;
+			} else if (file instanceof SyncFile) {
+				file.sync();
+				i++;
+			}
 		}
 	}
 

src/CAS.ts

@@ -0,0 +1,106 @@
+import type { LoginManager } from "./LoginManager";
+import type { FileInfo } from "./Relay";
+import type { FileInfoDAO, RelayManager } from "./RelayManager";
+import type { SharedFolder } from "./SharedFolder";
+import { customFetch } from "./customFetch";
+import { getMimeType } from "./mimetypes";
+import PocketBase from "pocketbase";
+
+declare const AUTH_URL: string;
+
+export class ContentAddressedStore {
+	private pb: PocketBase;
+
+	constructor(
+		private sharedFolder: SharedFolder,
+		private relayManager: RelayManager,
+		private loginManager: LoginManager,
+	) {
+		this.pb = new PocketBase(AUTH_URL, this.loginManager.authStore);
+	}
+
+	async listFiles(): Promise<FileInfo[]> {
+		return this.relayManager.fileInfo
+			.filter(
+				(fileInfo) => fileInfo.sharedFolder.id === this.sharedFolder.remote?.id,
+			)
+			.values();
+	}
+
+	async getByHash(hash: string): Promise<FileInfo | undefined> {
+		const local = this.relayManager.fileInfo.find(
+			(fileInfo) => fileInfo.synchash === hash,
+		);
+		if (local) {
+			return local;
+		}
+		try {
+			const records = await this.pb
+				?.collection("file_info")
+				.getFullList({ fetch: customFetch });
+			this.relayManager.store?.ingestBatch<FileInfo>(records);
+		} catch (e) {
+			// pass
+		}
+		return this.relayManager.fileInfo.find(
+			(fileInfo) => fileInfo?.synchash === hash,
+		);
+	}
+
+	async readFile(id: string): Promise<ArrayBuffer> {
+		const fileInfo = this.relayManager.fileInfo.get(id);
+		if (!fileInfo) throw new Error(`File not found: ${id}`);
+		const response = await fileInfo.getAttachment();
+		return response.arrayBuffer;
+	}
+
+	async writeFile(
+		item: Partial<FileInfoDAO>,
+		content: ArrayBuffer | null,
+	): Promise<FileInfo> {
+		if (!this.sharedFolder.remote) {
+			throw new Error("missing remote");
+		}
+		const blob =
+			content && item.name
+				? new Blob([content], { type: getMimeType(item.name) })
+				: null;
+		const fileData: Partial<FileInfoDAO<Blob>> = {
+			relay: this.sharedFolder.remote.relay.id,
+			shared_folder: this.sharedFolder.remote?.id,
+			guid: item.guid,
+			name: item.name,
+			synchash: item.synchash,
+			ctime: item.ctime,
+			mtime: item.mtime,
+			parent: item.parentId,
+			is_directory: item.isDirectory,
+			fileInfo: item.fileInfo,
+			synctime: item.synctime || 0,
+		};
+		if (blob) {
+			fileData["attachment"] = blob;
+		}
+
+		const record = item.fileInfo
+			? await this.pb.collection("file_info").update(item.fileInfo.id, fileData)
+			: await this.pb.collection("file_info").create(fileData);
+		const fileInfo = this.relayManager.store?.ingest<FileInfo>(record);
+		if (!fileInfo) throw new Error("Failed to create file");
+
+		return fileInfo;
+	}
+
+	async deleteFile(id: string): Promise<void> {
+		await this.pb.collection("file_info").delete(id);
+		this.relayManager.store?.cascade("file_info", id);
+	}
+
+	public destroy() {
+		this.pb.cancelAllRequests();
+		this.pb = null as any;
+		this.relayManager = null as any;
+		this.loginManager = null as any;
+		this.sharedFolder = null as any;
+	}
+}

src/Document.ts

@@ -5,12 +5,12 @@ import { HasProvider } from "./HasProvider";
 import { LoginManager } from "./LoginManager";
 import { S3Document, S3Folder, S3RemoteDocument } from "./S3RN";
 import { SharedFolder } from "./SharedFolder";
-import { curryLog } from "./debug";
 import type { TFile, Vault, TFolder } from "obsidian";
 import { DiskBuffer } from "./DiskBuffer";
 import type { Unsubscriber } from "./observable/Observable";
+import type { IFile } from "./IFile";
 
-export class Document extends HasProvider implements TFile {
+export class Document extends HasProvider implements IFile {
 	private _parent: SharedFolder;
 	private _persistence: IndexeddbPersistence;
 	_hasKnownPeers?: boolean;
@@ -28,18 +28,6 @@ export class Document extends HasProvider implements TFile {
 	_diskBuffer?: DiskBuffer;
 	offFolderStatusListener: Unsubscriber;
 
-	debug!: (message?: any, ...optionalParams: any[]) => void;
-	log!: (message?: any, ...optionalParams: any[]) => void;
-	warn!: (message?: any, ...optionalParams: any[]) => void;
-	error!: (message?: any, ...optionalParams: any[]) => void;
-
-	setLoggers(context: string) {
-		this.debug = curryLog(context, "debug");
-		this.log = curryLog(context, "log");
-		this.warn = curryLog(context, "warn");
-		this.error = curryLog(context, "error");
-	}
-
 	constructor(
 		path: string,
 		guid: string,
@@ -71,7 +59,6 @@ export class Document extends HasProvider implements TFile {
 			},
 		);
 
-		this.setLoggers(`[SharedDoc](${this.path})`);
 		try {
 			this._persistence = new IndexeddbPersistence(this.guid, this.ydoc);
 		} catch (e) {
@@ -98,6 +85,10 @@ export class Document extends HasProvider implements TFile {
 		this.updateStats();
 	}
 
+	async pull() {
+		await this.sharedFolder.backgroundSync.getDocument(this);
+	}
+
 	public get parent(): TFolder | null {
 		return this.tfile?.parent || null;
 	}

src/IFile.ts

@@ -0,0 +1,12 @@
+export interface IFile {
+	guid: string;
+	path: string;
+	move: (newPath: string) => void;
+	connect: () => void;
+	disconnect: () => void;
+	destroy: () => void;
+}
+
+export interface Hashable {
+	sha256: () => Promise<string>;
+}

src/LiveViews.ts

@@ -534,9 +534,11 @@ export class LiveViewManager {
 					});
 					views.push(view);
 				} else if (folder.ready) {
-					const doc = folder.getFile(viewFilePath, true, true, true);
-					const view = new LiveView(this, markdownView, doc);
-					views.push(view);
+					const file = folder.getFile(viewFilePath, true, true, true);
+					if (file instanceof Document) {
+						const view = new LiveView(this, markdownView, file);
+						views.push(view);
+					}
 				} else {
 					this.log(`Folder not ready, skipping views. folder=${folder.path}`);
 				}

src/LoginManager.ts

@@ -1,13 +1,13 @@
 "use strict";
 
-import { requestUrl } from "obsidian";
-import { User } from "./User";
 import PocketBase, {
 	BaseAuthStore,
 	type AuthProviderInfo,
 	type RecordAuthResponse,
 	type RecordModel,
 } from "pocketbase";
+import { requestUrl } from "obsidian";
+import { User } from "./User";
 import { RelayInstances, curryLog } from "./debug";
 import { Observable } from "./observable/Observable";
 
@@ -30,7 +30,7 @@ interface GoogleUser {
 export class LoginManager extends Observable<LoginManager> {
 	pb: PocketBase;
 	private openSettings: () => Promise<void>;
-	private authStore: LocalAuthStore;
+	authStore: LocalAuthStore;
 	user?: User;
 	resolve?: (code: string) => Promise<RecordAuthResponse<RecordModel>>;
 

src/Relay.ts

@@ -1,3 +1,4 @@
+import type { RequestUrlResponse } from "obsidian";
 import type { IObservable } from "./observable/Observable";
 import type { ObservableMap } from "./observable/ObservableMap";
 
@@ -9,6 +10,13 @@ interface Identified {
 interface Updatable<T> {
 	update(update: unknown): T;
 }
+interface HasAttachment {
+	attachmentUrl(): Promise<string>;
+	getAttachment(): Promise<RequestUrlResponse>;
+}
+interface Serializable {
+	toDict: () => any;
+}
 
 export interface RelayUser extends Identified, Updatable<RelayUser> {
 	id: string;
@@ -72,3 +80,30 @@ export interface RelaySubscription
 	quantity: number;
 	token: string;
 }
+
+export interface FileInfo
+	extends Identified,
+		Updatable<FileInfo>,
+		HasAttachment,
+		Serializable {
+	id: string;
+	guid: string;
+	relay: Relay;
+	parent: string | null;
+	sharedFolder: RemoteSharedFolder;
+	ctime: number;
+	mtime: number;
+	synchash: string;
+	synctime: number;
+	updated: string;
+	created: string;
+	type: string;
+	name: string;
+	deletedAt: number | null;
+	lastParentId: string | null;
+	isDirectory: boolean;
+}
+
+export interface FileInfoSend extends FileInfo {
+	attachment: null | Blob | File;
+}