import _ from 'lodash';
import https from 'https';
import path from 'path';
import fse from 'fs-extra';
import { glob } from 'glob';
import { MutationEvent, SanityAssetDocument, SanityClient as SanityClientType, SanityDocument, SanityDocumentStub } from '@sanity/client';
import type * as ContentSourceTypes from '@stackbit/types';
import type { DocumentVersion, DocumentVersionWithDocument, Field, Model, Schema, UserCommandSpawner } from '@stackbit/types';
import { getVersion as stackbitUtilsGetVersion } from '@stackbit/types';
import type * as SanityTypes from '@sanity/types';
import { deferWhileRunning, getPackageManager, omitByNil } from '@stackbit/utils';

import * as fetcher from './sanity-schema-fetcher';
import {
    DocumentHistory,
    DocumentHistoryMap,
    fetchDocumentForRevision,
    fetchDocumentRevision,
    fetchDocumentsHistory,
    fetchScheduledActions,
    fetchUsers,
    SanityClient,
    SanityUser,
    testToken
} from './sanity-api-client';
import { convertSchema, ModelContext, SchemaContext } from './sanity-schema-converter';
import {
    AssetContext,
    ContextualAsset,
    ContextualDocument,
    convertAndFilterScheduledActions,
    convertAssets,
    convertDocuments,
    ConvertDocumentsOptions,
    convertScheduledAction,
    DocumentContext,
    DRAFT_ID_PREFIX,
    getDraftObjectId,
    getPureObjectId,
    isDraftId
} from './sanity-document-converter';
import { isLocalizedModelField } from './utils';
import { convertUpdateOperation, localizedValue, mapUpdateOperationFieldToSanityValue } from './sanity-operation-converter';

export interface ContentSourceOptions {
    /**
     * The root path of the project. Used to resolve relative paths.
     */
    rootPath: string;

    /**
     * Sanity project id
     */
    projectId: string;

    /**
     * Sanity token with access to the project.
     * For local development, use a token with read and write access.
     */
    token: string;

    /**
     * Sanity dataset name. For exampe: 'production'
     */
    dataset: string;

    /**
     * Sanity studio url. Used for deep linking to documents.
     */
    studioUrl: string;

    /**
     * Sanity studio path. For example: 'studio'.
     * Usually the directory that contains your Sanity config file.
     */
    studioPath?: string;

    /**
     * The command to use to install the Sanity studio. For example: 'npm install'.
     * By default we detect the package manager and use it to install the studio.
     */
    studioInstallCommand?: string;

    /**
     * Glob pattern for Sanity schema files. Used to determine when a file changed schema has changed.
     */
    schemaGlob?: string;

    /**
     * Sanity query used to listen for document changes. For example: '*[!(_id in path("_.**"))]'
     */
    sanityQuery?: string;

    /**
     * The content change listener visibility. Defaults to 'transaction'.
     * 'transaction' - the listener will be notified of changes after the transaction is committed (fastest).
     * 'query' - the listener will be notified of changes when they are available for querying.
     */
    listenerVisibility?: 'transaction' | 'query';

    /**
     * When using Sanity Internationalized-Array plugin, specify the default language
     * that will be used by the Visual Editor. If not specified, the first language
     * in the `languages` will be selected.
     *
     * https://www.sanity.io/plugins/internationalized-array
     */
    defaultLocale?: string;
}

export type UserContext = {
    accessToken: string;
};

export type ContextualInitOptions = ContentSourceTypes.InitOptions<SchemaContext, DocumentContext, AssetContext, ModelContext>;
export type CacheWithContext = ContentSourceTypes.Cache<SchemaContext, DocumentContext, AssetContext, ModelContext>;
type ContentChanges = Required<ContentSourceTypes.ContentChanges<DocumentContext, AssetContext>>;

const ASSET_TYPES = ['sanity.imageAsset', 'cloudinary.asset', 'bynder.asset', 'aprimo.cdnasset'];
const SANITY_API_VERSION = '1';

export class SanityContentSource implements ContentSourceTypes.ContentSourceInterface<UserContext, SchemaContext, DocumentContext, AssetContext, ModelContext> {
    private readonly projectId: string;
    private readonly dataset: string;
    private readonly token: string;
    private readonly studioPath?: string;
    private readonly studioInstallCommand?: string;
    private readonly schemaGlob?: string;
    private readonly studioUrl: string;
    private readonly sanityQuery: string;
    private readonly rootPath: string;
    private readonly listenerVisibility?: 'transaction' | 'query';

    logger!: ContentSourceTypes.Logger;
    userLogger!: ContentSourceTypes.Logger;
    private localDev!: boolean;
    private runCommand!: ContentSourceTypes.CommandRunner;
    private userCommandSpawner?: UserCommandSpawner;
    private client!: SanityClientType;
    private contentChangeSubscription: any;
    private contentChangeSubscriptionInterval: any;
    private userMap: Record<string, SanityUser> = {};
    private cache!: CacheWithContext;
    private defaultLocale?: string;

    constructor(options: ContentSourceOptions) {
        this.projectId = options.projectId;
        this.dataset = options.dataset;
        this.token = options.token;
        this.studioUrl = options.studioUrl;
        this.sanityQuery = options.sanityQuery || '*[!(_id in path("_.**"))]';
        this.rootPath = options.rootPath;
        this.studioPath = options.studioPath;
        this.studioInstallCommand = options.studioInstallCommand;
        this.schemaGlob = options.schemaGlob;
        this.listenerVisibility = options.listenerVisibility;
        this.defaultLocale = options.defaultLocale;

        if (!this.rootPath) {
            throw new Error(`Required parameter 'rootPath' is missing.`);
        }

        this.convertListenerResult = deferWhileRunning(this.convertListenerResult, {
            thisArg: this,
            debounceDelay: 50,
            debounceMaxDelay: 500,
            argsResolver: ({ nextArgs, prevArgs }) => {
                const [events] = nextArgs;
                const [prevEvents] = prevArgs ?? [[]];
                return [[...prevEvents, ...events].filter(Boolean)] as typeof nextArgs;
            }
        });

        // Normalize paths to be absolute
        this.studioPath = !this.studioPath || path.isAbsolute(this.studioPath) ? this.studioPath : path.join(this.rootPath, this.studioPath);
    }

    async getVersion(): Promise<ContentSourceTypes.Version> {
        return stackbitUtilsGetVersion({ packageJsonPath: path.join(__dirname, '../package.json') });
    }

    getContentSourceType(): string {
        return 'sanity';
    }

    getProjectId(): string {
        return `${this.projectId}:${this.dataset}`;
    }

    getProjectEnvironment(): string {
        return this.dataset;
    }

    getProjectManageUrl(): string {
        return this.studioUrl;
    }

    async init({ logger, userLogger, localDev, runCommand, userCommandSpawner, cache }: ContextualInitOptions): Promise<void> {
        this.logger = logger.createLogger({ label: 'cms-sanity' });
        this.userLogger = userLogger.createLogger({ label: 'cms-sanity' });
        this.localDev = localDev;
        this.runCommand = runCommand;
        this.userCommandSpawner = userCommandSpawner;
        this.cache = cache;

        this.logger.debug('init', { localDev });

        this.client = new SanityClient({
            projectId: this.projectId,
            dataset: this.dataset,
            token: this.token,
            useCdn: false,
            apiVersion: SANITY_API_VERSION
        });

        await this.validateConfig();

        if (!localDev) {
            await this.installStudio();
        }

        await this.reset();
    }

    async reset() {
        this.logger.debug('reset');
    }

    async destroy() {
        this.logger.debug('destroy');
    }

    async validateConfig() {
        this.logger.debug('Validating config...');
        if (this.studioPath && !(await fse.pathExists(this.studioPath))) {
            throw new Error(`Can't find Sanity Studio in ${this.studioPath}. Verify that the studio path is pointing to the right place.`);
        }
        if (_.isEmpty(this.token)) {
            throw new Error(`Please provide a Sanity token.`);
        }
        try {
            await this.client.fetch(`*[_type == 'system.group'] {_id}`);
        } catch (err: any) {
            if (err?.response?.body?.error === 'Dataset not found') {
                throw new Error(`Can't find specified Sanity dataset '${this.dataset}'. Verify that the dataset is pointing to an existing Sanity dataset.`);
            } else {
                throw new Error(`Can't connect to Sanity. Verify that the token has access to the Sanity project defined.`);
            }
        }
    }

    async installStudio() {
        if (this.studioInstallCommand) {
            this.userLogger.info('Installing Sanity Studio using custom command');
            await this.runCommand(this.studioInstallCommand, [], {
                shell: true,
                cwd: this.studioPath ?? this.rootPath,
                logger: this.userLogger.createLogger({ label: 'sanity-studio-install' })
            });
        } else {
            if (!this.studioPath || this.studioPath === this.rootPath) {
                this.logger.debug('Studio path is the same as root path. Skipping studio installation.');
                return;
            }
            const packageManager = await getPackageManager(this.studioPath);
            if (!packageManager) {
                throw new Error("No package manager detected. Can't install Sanity Studio.");
            }
            this.userLogger.info(`Installing Sanity Studio using ${packageManager.name}`);
            await this.runCommand(packageManager.cmd, packageManager.args, {
                env: packageManager.env,
                cwd: this.studioPath,
                logger: this.userLogger.createLogger({ label: 'sanity-studio-install' })
            });
        }
    }

    startWatchingContentUpdates() {
        this.logger.debug('startWatchingContentUpdates');

        const createSanityListener = () => {
            //TODO Sanity types for listener visibility are broken in this @sanity/client version
            return this.client.listen(this.sanityQuery, {}, { visibility: this.listenerVisibility === 'query' ? 'query' : undefined }).subscribe({
                next: async (event) => {
                    if (event.type !== 'mutation') {
                        return;
                    }
                    if (_.startsWith(event.documentId, '_.') || event.identity === '<system>' || _.startsWith(event.result?.resultType, 'system.')) {
                        return;
                    }
                    this.logger.debug('SanityListener: content changed', {
                        event: event?.transition,
                        transactionId: event?.transactionId,
                        eventId: event?.eventId,
                        documentId: event?.documentId
                    });
                    await this.convertListenerResult([event]).catch((err) => {
                        this.logger.error('SanityListener: convertListenerResult threw an error', err);
                    });
                    this.logger.debug('SanityListener: content changed handler done', {
                        event: event?.transition,
                        transactionId: event?.transactionId,
                        eventId: event?.eventId,
                        documentId: event?.documentId
                    });
                },
                error: (error) => {
                    this.logger.error('SanityListener: threw error', error);
                },
                complete: () => {
                    this.logger.debug('SanityListener: complete received');
                }
            });
        };

        if (this.contentChangeSubscription) {
            this.stopWatchingContentUpdates();
        }

        this.contentChangeSubscription = createSanityListener();
        this.contentChangeSubscriptionInterval = setInterval(() => {
            const newListener = createSanityListener();
            this.contentChangeSubscription.unsubscribe();
            this.contentChangeSubscription = newListener;
        }, 1000 * 60 * 20);
    }

    stopWatchingContentUpdates() {
        this.logger.debug('stopWatchingContentUpdates');
        if (this.contentChangeSubscription) {
            this.logger.debug('stopping content change listener');
            this.contentChangeSubscription.unsubscribe();
            this.contentChangeSubscription = null;
            clearInterval(this.contentChangeSubscriptionInterval);
            this.contentChangeSubscriptionInterval = null;
        }
    }

    private async convertListenerResult(events: MutationEvent<any>[]): Promise<void> {
        const result: ContentChanges = {
            documents: [],
            assets: [],
            scheduledActions: [],
            deletedDocumentIds: [],
            deletedAssetIds: [],
            deletedScheduledActionIds: []
        };
        const logContext = {
            event: events?.map((event) => event?.transition)?.join(','),
            transactionId: events?.map((event) => event?.transactionId)?.join(','),
            eventId: events?.map((event) => event?.eventId)?.join(','),
            documentId: events?.map((event) => event?.documentId)?.join(',')
        };
        this.logger.debug('SanityListener: content changed, convert started', logContext);

        // fetch document history once for all events
        const documentIds = _.uniq(
            events.filter((event) => event.transition !== 'disappear' && event.result && isDraftId(event.documentId)).map((event) => event.documentId)
        ).filter(Boolean) as string[];
        const documentsHistory = await this.getDocumentsHistory(documentIds);

        // fetch updated scheduled actions after a second delay to allow sanity to update the status of any executed scheduled action
        const pureDocumentIds = _.uniq(events.map((event) => getPureObjectId(event.documentId))).filter(Boolean);
        setTimeout(async () => {
            try {
                const sanitySchedules = await fetchScheduledActions({ projectId: this.projectId, dataset: this.dataset }, this.client, {
                    documentIds: pureDocumentIds
                });
                const remoteScheduledActions = convertAndFilterScheduledActions(sanitySchedules);
                const cachedScheduledActions = pureDocumentIds.flatMap((documentId) => this.cache.getScheduledActionsForDocumentId(documentId));
                const updatedActions = _.differenceWith(remoteScheduledActions, cachedScheduledActions, _.isEqual);
                if (updatedActions.length) {
                    this.logger.debug('SanityListener: Scheduled actions changed, updating', { updatedSchedulesCount: updatedActions.length });
                    this.cache.updateContent({ scheduledActions: updatedActions });
                }
            } catch (err) {
                this.logger.debug('SanityListener: Failed to fetch Scheduled actions for updated documents', { ...logContext, error: err });
            }
        }, 1000);

        // save local cache of documents and assets to be reused between events
        const documentMap: Record<string, ContextualDocument> = {};
        const assetMap: Record<string, ContextualAsset> = {};

        for (const event of events) {
            const pureObjectId = getPureObjectId(event.documentId);
            const document = documentMap[pureObjectId] || this.cache.getDocumentById(pureObjectId);
            const asset = this.cache.getAssetById(pureObjectId);
            const isDraft = isDraftId(event.documentId);

            // cleanup deleted ids - either prevents duplicates or removes ids that are no longer deleted
            const existingDeletedDocumentIdIndex = result.deletedDocumentIds.indexOf(pureObjectId);
            if (existingDeletedDocumentIdIndex >= 0) {
                result.deletedDocumentIds = result.deletedDocumentIds.splice(existingDeletedDocumentIdIndex, 1);
            }
            const existingDeletedAssetIdIndex = result.deletedAssetIds.indexOf(pureObjectId);
            if (existingDeletedAssetIdIndex >= 0) {
                result.deletedAssetIds = result.deletedAssetIds.splice(existingDeletedAssetIdIndex, 1);
            }

            if (event.transition === 'disappear') {
                if (document) {
                    const { context } = document;
                    if ((isDraft && !context.publishedDocument) || (!isDraft && !context.draftDocument)) {
                        // notify deleted only if already in content cache
                        if (this.cache.getDocumentById(pureObjectId)) {
                            result.deletedDocumentIds.push(pureObjectId);
                        }
                        // remove now-deleted document from result
                        if (documentMap[pureObjectId]) {
                            delete documentMap[pureObjectId];
                        }
                    } else {
                        const documents = [isDraft ? context.publishedDocument! : context.draftDocument!];
                        const [convertedDocuments] = this.convertDocuments({
                            documents,
                            getModelByName: this.cache.getModelByName,
                            studioUrl: this.studioUrl,
                            documentsHistory
                        });
                        if (convertedDocuments) {
                            documentMap[pureObjectId] = convertedDocuments;
                        }
                    }
                } else if (asset) {
                    result.deletedAssetIds.push(pureObjectId);
                    if (assetMap[pureObjectId]) {
                        delete assetMap[pureObjectId];
                    }
                }
            }

            if (event.result) {
                const publishedDocument = isDraft ? document?.context.publishedDocument : event.result;
                const draftDocument = isDraft ? event.result : document?.context.draftDocument;
                const documents = [publishedDocument, draftDocument].filter(Boolean) as SanityDocument[];
                if (ASSET_TYPES.includes(event.result._type)) {
                    const convertedAssets = convertAssets({ assets: documents, documentsHistory });
                    assetMap[pureObjectId] = convertedAssets[0]!;
                } else {
                    const [convertedDocuments] = this.convertDocuments({
                        documents: documents,
                        getModelByName: this.cache.getModelByName,
                        studioUrl: this.studioUrl,
                        documentsHistory
                    });
                    if (convertedDocuments) {
                        documentMap[pureObjectId] = convertedDocuments;
                    }
                }
            }
        }

        result.documents = Object.values(documentMap);
        result.assets = Object.values(assetMap);

        this.logger.debug('SanityListener: content changed, convert done', logContext);
        return this.cache.updateContent(result);
    }

    private convertVersionsFromDocumentHistory(versions: DocumentHistory[]): DocumentVersion[] {
        return versions.map((version) => {
            return {
                id: version.id,
                documentId: getPureObjectId(version.documentIDs[0]!),
                srcType: this.getContentSourceType(),
                srcProjectId: this.projectId,
                createdAt: version.timestamp,
                createdBy: this.userMap[version.author]?.email
            };
        });
    }

    private convertVersionForDocument(version: DocumentHistory, document: ContextualDocument): DocumentVersionWithDocument {
        const [documentVersion] = this.convertVersionsFromDocumentHistory([version]);
        if (!documentVersion) {
            throw new Error('Document version could not be converted');
        }
        return {
            ...documentVersion,
            document
        };
    }

    async getSanitySchema(): Promise<{ models: SanityTypes.SchemaTypeDefinition[] }> {
        let sanitySchema;
        try {
            sanitySchema = await fetcher.spawnFetchSchema({
                studioPath: this.studioPath ?? this.rootPath,
                repoPath: this.rootPath,
                spawnRunner: this.userCommandSpawner,
                logger: this.logger
            });
        } catch (err) {
            if (!_.isEmpty(err) && _.isString(err)) {
                this.userLogger.info(err);
            }
            this.userLogger.error('Error fetching Sanity schema');
            throw err;
        }
        return sanitySchema;
    }

    async getSchema(): Promise<Schema<SchemaContext, ModelContext>> {
        this.logger.debug('getSchema');
        const sanitySchema = await this.getSanitySchema();
        const { models, locales } = convertSchema({
            schema: sanitySchema,
            logger: this.logger,
            defaultLocale: this.defaultLocale
        });
        return {
            models,
            locales,
            context: null
        };
    }

    async getDocumentsHistory(documentIds: string[]): Promise<DocumentHistoryMap> {
        this.logger.debug('getDocumentsHistory started', documentIds.length);

        const draftDocumentIds = documentIds.filter((documentId) => isDraftId(documentId));
        const historyData = await fetchDocumentsHistory({ documentIds: draftDocumentIds, dataset: this.dataset, client: this.client });

        const notCachedDocAuthors = historyData.reduce((acc: string[], doc: DocumentHistory) => {
            const user = this.userMap[doc.author];
            if (!user && !acc.includes(doc.author) && doc.author !== 'system' && doc.author !== '<system>') {
                acc.push(doc.author);
            }
            return acc;
        }, []);

        if (notCachedDocAuthors.length) {
            try {
                const authors = await fetchUsers(notCachedDocAuthors, this.client);
                for (const author of authors) {
                    this.userMap[author.id] = author;
                }
            } catch (err) {
                this.logger.error('getDocumentsHistory error fetching authors', err);
                this.userLogger.warn('Error fetching users from Sanity. Make sure the Sanity token has the appropriate permissions.');
            }
        }

        this.logger.debug('getDocumentsHistory done fetch');

        return draftDocumentIds.reduce((acc: DocumentHistoryMap, documentId: string) => {
            const docHistory = historyData.filter(({ documentIDs }: DocumentHistory) => documentIDs.includes(documentId));
            const docHistoryWithAuthor = [];

            for (const history of docHistory) {
                const author = this.userMap[history.author];

                if (!author) {
                    this.logger.debug(`getDocumentsHistory author is missing in userMap`, { authorId: history.author });
                    continue;
                }

                if (author.email) {
                    docHistoryWithAuthor.push({
                        ...history,
                        author: author.email
                    });
                }
            }

            if (docHistoryWithAuthor.length) {
                acc[documentId] = docHistoryWithAuthor;
            }

            return acc;
        }, {});
    }

    // centralized place that handles document conversion to allow easy content source extension
    convertDocuments(options: ConvertDocumentsOptions) {
        return convertDocuments(options);
    }

    async getDocuments(): Promise<ContextualDocument[]> {
        this.logger.debug('getDocuments');
        const documents: SanityDocument[] = await this.client.fetch(this.sanityQuery);
        this.logger.debug(`fetched ${documents.length} entries from project ${this.projectId} and dataset ${this.dataset}`);
        const documentsHistory = await this.getDocumentsHistory(documents.map((document) => document._id));

        return this.convertDocuments({
            documents: documents.filter((document: SanityDocument) => !ASSET_TYPES.includes(document._type)),
            getModelByName: this.cache.getModelByName,
            documentsHistory,
            studioUrl: this.studioUrl
        });
    }

    async getAssets() {
        const documents: SanityDocument[] = await this.client.fetch('*[_type in $types]', { types: ASSET_TYPES });
        this.logger.debug(`fetched ${documents.length} assets from project ${this.projectId} and dataset ${this.dataset}`);
        const documentsHistory = await this.getDocumentsHistory(documents.map((document) => document._id));

        return convertAssets({
            assets: documents.filter((document: SanityDocument) => ASSET_TYPES.includes(document._type)),
            documentsHistory
        });
    }

    async hasAccess({ userContext }: { userContext?: UserContext }): Promise<{ hasConnection: boolean; hasPermissions: boolean }> {
        if (!this.localDev) {
            if (!userContext?.accessToken) {
                return {
                    hasConnection: false,
                    hasPermissions: false
                };
            }
            let tokenExpired = true;
            try {
                const testTokenResult = await testToken(userContext.accessToken);
                const accessTokenExpiresAt = testTokenResult?.accessTokenExpiresAt;
                if (accessTokenExpiresAt) {
                    const expiresAt = new Date(accessTokenExpiresAt);
                    tokenExpired = new Date().getTime() - expiresAt.getTime() >= 0;
                }
            } catch (e) {
                // pass
            }
            if (tokenExpired) {
                return {
                    hasConnection: false,
                    hasPermissions: false
                };
            }
        }
        try {
            const userClient = this.getApiClientForUser({ userContext });
            await userClient.fetch(`*[_type == 'system.group'] {_id}`);
            return {
                hasConnection: true,
                hasPermissions: true
            };
        } catch (error) {
            this.logger.debug('Sanity: failed to access project', { error });
            return {
                hasConnection: true,
                hasPermissions: false
            };
        }
    }

    async onFilesChange({ updatedFiles }: { updatedFiles: string[] }) {
        let invalidateSchema = false;

        if (this.studioPath) {
            const relStudioPath = path.relative(this.rootPath, this.studioPath);
            const sanityPackageJson = path.join(relStudioPath, 'package.json');
            const needsStudioInstall = _.some(updatedFiles, (updatedFile) => updatedFile === sanityPackageJson);

            invalidateSchema = _.some(updatedFiles, (updatedFile) => updatedFile.startsWith(relStudioPath));

            if (needsStudioInstall) {
                this.logger.debug('Sanity Studio package.json changed, reinstalling...');
                await this.installStudio();
            }
        }

        if (this.schemaGlob) {
            const existingSchemaFiles = await glob(this.schemaGlob, { cwd: this.rootPath, absolute: false });
            invalidateSchema = !_.isEmpty(_.intersection(updatedFiles, existingSchemaFiles));
        }

        return { invalidateSchema };
    }

    async createDocument({
        updateOperationFields,
        model,
        locale,
        userContext
    }: {
        updateOperationFields: Record<string, ContentSourceTypes.UpdateOperationField>;
        model: Model<ModelContext>;
        locale?: string;
        userContext?: UserContext;
    }): Promise<{ documentId: string }> {
        const sanityDocument: SanityDocumentStub = {
            _id: DRAFT_ID_PREFIX,
            _type: model.name
        };

        try {
            _.forEach(updateOperationFields, (updateOperationField, fieldName) => {
                const childModelField = _.find(model.fields, (field: Field) => field.name === fieldName);
                if (!childModelField) {
                    throw new Error(`No model field found for field: ${fieldName}`);
                }
                const value = mapUpdateOperationFieldToSanityValue({
                    updateOperationField,
                    getModelByName: this.cache.getModelByName,
                    modelField: childModelField,
                    rootModel: model,
                    modelFieldPath: [fieldName],
                    locale
                });
                if (isLocalizedModelField(childModelField)) {
                    _.set(sanityDocument, fieldName, [
                        localizedValue({
                            value,
                            model,
                            modelFieldPath: [fieldName],
                            locale
                        })
                    ]);
                } else {
                    _.set(sanityDocument, fieldName, value);
                }
            });
            const userClient = this.getApiClientForUser({ userContext });
            const result = await userClient.create(sanityDocument);
            const pureObjectId = getPureObjectId(result._id);
            return { documentId: pureObjectId };
        } catch (error: any) {
            this.logger.error('Error creating document', error);
            throw new Error(`Error creating document. ${error.message}`);
        }
    }

    async updateDocument({
        document,
        operations,
        userContext
    }: {
        document: ContextualDocument;
        operations: ContentSourceTypes.UpdateOperation[];
        userContext?: UserContext;
    }): Promise<void> {
        this.logger.debug('updateDocument');

        const documentId = document.id;
        const userClient = this.getApiClientForUser({ userContext });
        const draftObjectId = getDraftObjectId(documentId);

        const sanityDocument = (document.context.draftDocument || document.context.publishedDocument)!;
        const modelName = document.modelName;
        const model = this.cache.getModelByName(modelName);
        try {
            if (!model) {
                throw new Error(`Could not find document model '${modelName}'.`);
            }
            const transaction = _.reduce(
                operations,
                (transaction, operation) => {
                    const patchObject = convertUpdateOperation({
                        operation,
                        sanityDocument,
                        getModelByName: this.cache.getModelByName,
                        model
                    });
                    return transaction.patch(draftObjectId, patchObject);
                },
                userClient.transaction().createIfNotExists({
                    ...sanityDocument,
                    _id: draftObjectId
                })
            );
            await transaction.commit({ visibility: this.listenerVisibility === 'query' ? 'sync' : 'async', returnDocuments: true });
        } catch (error: any) {
            this.logger.error(`Error updating document ${document.id}`, error);
            throw new Error(`Error updating document ${document.id}. ${error.message}`);
        }
    }

    async deleteDocument({ document, userContext }: { document: ContextualDocument; userContext?: UserContext }): Promise<void> {
        const userClient = this.getApiClientForUser({ userContext });
        const draftObjectId = getDraftObjectId(document.id);
        const pureObjectId = getPureObjectId(document.id);
        await Promise.all([userClient.delete(pureObjectId), userClient.delete(draftObjectId)]);
    }

    async uploadAsset({
        url,
        base64,
        fileName,
        mimeType,
        userContext
    }: {
        url?: string;
        base64?: string;
        fileName: string;
        mimeType: string;
        userContext?: UserContext;
    }): Promise<ContextualAsset> {
        const userClient = this.getApiClientForUser({ userContext });
        const assetType = mimeType.startsWith('image/') ? 'image' : 'file';
        let uploadResult: SanityAssetDocument;
        if (url) {
            uploadResult = await new Promise((resolve, reject) => {
                https.get(url, (downloadStream: any) => {
                    userClient.assets.upload(assetType, downloadStream, { filename: fileName }).then(resolve).catch(reject);
                });
            });
        } else {
            const buffer = Buffer.from(base64!, 'base64');
            uploadResult = await userClient.assets.upload(assetType, buffer, { filename: fileName });
        }
        const documentsHistory = await this.getDocumentsHistory([uploadResult._id]);
        const assets = convertAssets({ assets: [uploadResult!], documentsHistory });
        return assets[0]!;
    }

    async validateDocuments({
        documents,
        assets,
        locale,
        userContext
    }: {
        documents: ContextualDocument[];
        assets: ContextualAsset[];
        locale?: string;
        userContext?: UserContext;
    }): Promise<{ errors: ContentSourceTypes.ValidationError[] }> {
        return { errors: [] };
    }

    async publishDocuments({
        documents,
        assets,
        userContext
    }: {
        documents: ContextualDocument[];
        assets: ContextualAsset[];
        userContext?: UserContext;
    }): Promise<void> {
        const userClient = this.getApiClientForUser({ userContext });
        const publishedDocumentIds = documents.map((document) => getPureObjectId(document.id));
        const draftDocumentIds = publishedDocumentIds.map((documentId) => getDraftObjectId(documentId));

        const query = '*[ _id in $documentIds ]';
        const sanityDocuments: SanityDocument[] = await userClient.fetch(query, { documentIds: [...draftDocumentIds, ...publishedDocumentIds] });
        const draftDocuments = sanityDocuments.filter((document) => isDraftId(document._id));

        const transaction = _.reduce(
            draftDocuments,
            (transaction, document) => {
                const documentId = document._id;
                const publishedDocument = _.find(sanityDocuments, { _id: getPureObjectId(documentId) });

                if (publishedDocument) {
                    // borrowed form Sanity folks - https://github.com/sanity-io/sanity/blob/c3c875ed51bf49ebceedc40abe50ad17ccbf489b/packages/%40sanity/desk-tool/src/pane/DocumentPane.js#L860
                    // Hack until other mutations support revision locking
                    transaction = transaction.patch(publishedDocument._id, {
                        unset: ['_reserved_prop_'],
                        ifRevisionID: publishedDocument._rev
                    });
                }

                // you should omit it to be sure it get's the right timestamp from the server
                const documentForPublishing = _.omit(document, ['_updatedAt']) as SanityDocumentStub;

                return transaction
                    .createOrReplace({
                        ...documentForPublishing,
                        _id: getPureObjectId(documentId)
                    })
                    .delete(documentId);
            },
            userClient.transaction()
        );

        await transaction.commit();
    }

    async getScheduledActions(): Promise<ContentSourceTypes.ScheduledAction[]> {
        this.logger.debug('getScheduledPublishes');
        const sanitySchedules = await fetchScheduledActions({ projectId: this.projectId, dataset: this.dataset }, this.client);
        return convertAndFilterScheduledActions(sanitySchedules);
    }

    async createScheduledAction({
        documentIds,
        name,
        action,
        executeAt,
        userContext
    }: {
        documentIds: string[];
        name: string;
        action: ContentSourceTypes.ScheduledActionActionType;
        executeAt: string;
        userContext?: UserContext;
    }): Promise<{ newScheduledActionId: string }> {
        this.logger.debug('createScheduledAction');
        const userClient = this.getApiClientForUser({ userContext });
        const sanitySchedule = await userClient.request({
            method: 'POST',
            uri: `/schedules/${this.projectId}/${this.dataset}`,
            body: {
                documents: documentIds.map((docId) => ({ documentId: docId })),
                executeAt: executeAt,
                name: name
            }
        });
        const schedule = convertScheduledAction(sanitySchedule);
        await this.cache.updateContent({ scheduledActions: [schedule] });
        this.logger.debug('createScheduledAction - Create success', { originalSchedule: sanitySchedule, schedule });
        return { newScheduledActionId: schedule.id };
    }

    async cancelScheduledAction({
        scheduledActionId,
        userContext
    }: {
        scheduledActionId: string;
        userContext?: UserContext;
    }): Promise<{ cancelledScheduledActionId: string }> {
        this.logger.debug('cancelScheduledAction');
        const userClient = this.getApiClientForUser({ userContext });
        const response = await userClient
            .request({
                method: 'PATCH',
                uri: `/schedules/${this.projectId}/${this.dataset}/${scheduledActionId}`,
                body: {
                    state: 'cancelled'
                }
            })
            .then(() =>
                userClient.request({
                    method: 'GET',
                    uri: `/schedules/${this.projectId}/${this.dataset}/${scheduledActionId}`
                })
            );
        const sanitySchedule = response.schedules[0];
        const schedule = convertScheduledAction(sanitySchedule);
        await this.cache.updateContent({ scheduledActions: [schedule] });
        this.logger.debug('cancelScheduledAction - Cancel success', { originalSchedule: sanitySchedule, schedule });
        return { cancelledScheduledActionId: schedule.id };
    }

    async updateScheduledAction({
        scheduledActionId,
        documentIds,
        name,
        executeAt,
        userContext
    }: {
        scheduledActionId: string;
        documentIds?: string[];
        name?: string;
        executeAt?: string;
        userContext?: UserContext;
    }): Promise<{ updatedScheduledActionId: string }> {
        this.logger.debug('updateScheduledAction');
        const updateObj = omitByNil({
            documents: documentIds?.map((documentId) => ({ documentId })),
            name,
            executeAt
        });
        const userClient = this.getApiClientForUser({ userContext });
        const response = await userClient
            .request({
                method: 'PATCH',
                uri: `/schedules/${this.projectId}/${this.dataset}/${scheduledActionId}`,
                body: updateObj
            })
            .then(() =>
                userClient.request({
                    method: 'GET',
                    uri: `/schedules/${this.projectId}/${this.dataset}/${scheduledActionId}`
                })
            );
        const sanitySchedule = response.schedules[0];
        const schedule = convertScheduledAction(sanitySchedule);
        await this.cache.updateContent({ scheduledActions: [schedule] });
        this.logger.debug('updateScheduledAction - Update success', { originalSchedule: sanitySchedule, schedule });
        return { updatedScheduledActionId: schedule.id };
    }

    getApiClientForUser({ userContext }: { userContext?: UserContext }): SanityClientType {
        if (this.localDev) {
            return this.client;
        }
        const userAccessToken = userContext?.accessToken;
        if (!userAccessToken) {
            throw new Error(`Permissions error: user does not have an access token for project: '${this.projectId}'.`);
        }
        return new SanityClient({
            projectId: this.projectId,
            dataset: this.dataset,
            token: userAccessToken,
            useCdn: false,
            apiVersion: SANITY_API_VERSION
        });
    }

    async getDocumentVersions({ documentId }: { documentId: string }): Promise<{ versions: DocumentVersion[] }> {
        this.logger.debug('getDocumentVersions', { documentId });

        const documentHistory = await fetchDocumentsHistory({
            documentIds: [documentId, getDraftObjectId(documentId)],
            dataset: this.dataset,
            client: this.client,
            limitTime: false
        });

        if (!documentHistory) {
            return { versions: [] };
        }

        return { versions: this.convertVersionsFromDocumentHistory(documentHistory) };
    }

    async getDocumentForVersion({ documentId, versionId }: { documentId: string; versionId: string }): Promise<{ version: DocumentVersionWithDocument }> {
        this.logger.debug('getDocumentForVersion', { documentId, versionId });

        const draftDocumentId = getDraftObjectId(documentId);
        const [version, document] = await Promise.all([
            fetchDocumentRevision({ documentId, draftDocumentId, versionId, dataset: this.dataset, client: this.client }),
            fetchDocumentForRevision({ documentId, draftDocumentId, versionId, dataset: this.dataset, client: this.client })
        ]);
        const [contextualDocument] = this.convertDocuments({
            documents: [document],
            getModelByName: this.cache.getModelByName,
            studioUrl: this.studioUrl
        });

        if (!contextualDocument) {
            throw new Error(`Could not get document ${documentId} for revision ${versionId}`);
        }

        return { version: this.convertVersionForDocument(version, contextualDocument) };
    }
}