reconcile/plugin/src/services/sync_service.ts

import * as lib from "../../../backend/sync_lib/pkg/sync_lib.js";

import createClient, { Client } from "openapi-fetch";
import type { components, paths } from "./types"; // generated by openapi-typescript
import { Logger } from "src/logger";
import { Database } from "src/database/database";
import { SyncSettings } from "src/database/sync-settings";
import {
	VaultUpdateId,
	RelativePath,
	DocumentId,
} from "src/database/document-metadata";
import PQueue from "p-queue";

export class SyncService {
	private promiseQueue: PQueue;
	private client: Client<paths>;

	public constructor(private database: Database) {
		this.createClient(database.getSettings());
		this.promiseQueue = new PQueue({
			concurrency: database.getSettings().uploadConcurrency,
		});

		database.addOnSettingsChangeHandlers((s) => {
			this.createClient(s);
			this.promiseQueue.concurrency = s.uploadConcurrency;
		});
	}

	private createClient(settings: SyncSettings) {
		this.client = createClient<paths>({
			baseUrl: settings.remoteUri,
		});
	}

	private enqueue<T>(fn: () => Promise<T>): Promise<T> {
		return this.promiseQueue.add(fn) as Promise<T>;
	}

	public async ping(): Promise<components["schemas"]["PingResponse"]> {
		const response = await this.enqueue(() =>
			this.client.GET("/ping", {
				params: {
					header: {
						authorization:
							"Bearer " + this.database.getSettings().token,
					},
				},
			})
		);

		Logger.getInstance().debug(
			"Ping response: " + JSON.stringify(response.data)
		);

		if (!response.data) {
			throw new Error(`Failed to ping server: ${response.error}`);
		}

		return response.data;
	}

	public async create({
		relativePath,
		contentBytes,
		createdDate,
	}: {
		relativePath: RelativePath;
		contentBytes: Uint8Array;
		createdDate: Date;
	}): Promise<components["schemas"]["DocumentVersion"]> {
		const response = await this.enqueue(() =>
			this.client.POST("/vaults/{vault_id}/documents", {
				params: {
					path: {
						vault_id: this.database.getSettings().vaultName,
					},
					header: {
						authorization:
							"Bearer " + this.database.getSettings().token,
					},
				},
				body: {
					contentBase64: lib.bytes_to_base64(contentBytes),
					createdDate: createdDate.toISOString(),
					relativePath,
				},
			})
		);

		if (!response.data) {
			throw new Error(`Failed to create document: ${response.error}`);
		}

		Logger.getInstance().debug(
			"Created document " + JSON.stringify(response.data)
		);

		return response.data;
	}

	public async put({
		parentVersionId,
		documentId,
		relativePath,
		contentBytes,
		createdDate,
	}: {
		parentVersionId: VaultUpdateId;
		documentId: DocumentId;
		relativePath: RelativePath;
		contentBytes: Uint8Array;
		createdDate: Date;
	}): Promise<components["schemas"]["DocumentVersion"]> {
		const response = await this.enqueue(() =>
			this.client.PUT("/vaults/{vault_id}/documents/{document_id}", {
				params: {
					path: {
						vault_id: this.database.getSettings().vaultName,
						document_id: documentId,
					},
					header: {
						authorization:
							"Bearer " + this.database.getSettings().token,
					},
				},
				body: {
					parentVersionId,
					contentBase64: lib.bytes_to_base64(contentBytes),
					createdDate: createdDate.toISOString(),
					relativePath,
				},
			})
		);

		if (!response.data) {
			throw new Error(`Failed to update document: ${response.error}`);
		}

		Logger.getInstance().debug(
			"Updated document " + JSON.stringify(response.data)
		);

		return response.data;
	}

	public async delete({
		documentId,
		relativePath,
		createdDate,
	}: {
		documentId: DocumentId;
		relativePath: RelativePath;
		createdDate: Date;
	}): Promise<void> {
		const response = await this.enqueue(() =>
			this.client.DELETE("/vaults/{vault_id}/documents/{document_id}", {
				params: {
					path: {
						vault_id: this.database.getSettings().vaultName,
						document_id: documentId,
					},
					header: {
						authorization:
							"Bearer " + this.database.getSettings().token,
					},
				},
				body: {
					createdDate: createdDate.toISOString(),
					relativePath,
				},
			})
		);

		if (response.error) {
			throw new Error(`Failed to delete document`);
		}

		Logger.getInstance().debug(
			"Updated document " + JSON.stringify(response.data)
		);

		return response.data;
	}

	public async get({
		documentId,
	}: {
		documentId: DocumentId;
	}): Promise<components["schemas"]["DocumentVersion"]> {
		const response = await this.enqueue(() =>
			this.client.GET("/vaults/{vault_id}/documents/{document_id}", {
				params: {
					path: {
						vault_id: this.database.getSettings().vaultName,
						document_id: documentId,
					},
					header: {
						authorization:
							"Bearer " + this.database.getSettings().token,
					},
				},
			})
		);

		if (!response.data) {
			throw new Error(`Failed to get document: ${response.error}`);
		}

		Logger.getInstance().debug(
			"Get document " + JSON.stringify(response.data)
		);

		return response.data;
	}

	public async getAll(
		since?: VaultUpdateId
	): Promise<components["schemas"]["FetchLatestDocumentsResponse"]> {
		const response = await this.enqueue(() =>
			this.client.GET("/vaults/{vault_id}/documents", {
				params: {
					path: {
						vault_id: this.database.getSettings().vaultName,
					},
					header: {
						authorization:
							"Bearer " + this.database.getSettings().token,
					},
					query: {
						since_update_id: since,
					},
				},
			})
		);

		if (!response.data) {
			throw new Error(`Failed to get documents: ${response.error}`);
		}

		Logger.getInstance().debug(
			"Get document " + JSON.stringify(response.data)
		);

		return response.data;
	}
}