<commit_before>import express from 'express';
import * as path from 'path';
const { graphqlHTTP } = require('express-graphql');
const cors = require('cors');
import { PubSub } from 'graphql-subscriptions';
import {
    buildSchema,
    execute,
    subscribe
} from 'graphql';
import { SubscriptionServer } from 'subscriptions-transport-ws';
import { createServer } from 'http';

import {
    GRAPHQL_PORT,
    GRAPHQL_PATH,
    GRAPHQL_SUBSCRIPTION_PORT,
    GRAPHQL_SUBSCRIPTION_PATH,
    graphQLGenerationInput,
    JWT_BEARER_TOKEN
} from '../shared';

import {
    graphQLSchemaFromRxSchema
} from 'rxdb/plugins/replication-graphql';

import {
    lastOfArray
} from 'rxdb';

function log(msg) {
    const prefix = '# GraphQL Server: ';
    if (typeof msg === 'string') {
        console.log(prefix + msg);
    } else {
        console.log(prefix + JSON.stringify(msg, null, 2));
    }
}

function sortByUpdatedAtAndPrimary(a, b) {
    if (a.updatedAt > b.updatedAt) return 1;
    if (a.updatedAt < b.updatedAt) return -1;

    if (a.updatedAt === b.updatedAt) {
        if (a.id > b.id) return 1;
        if (a.id < b.id) return -1;
        else return 0;
    }
}


/**
 * Returns true if the request is authenticated
 * throws if not.
 * In a real world app you would parse and validate the bearer token.
 * @link https://graphql.org/graphql-js/authentication-and-express-middleware/
 */
export function authenticateRequest(request) {
    const authHeader = request.header('authorization');
    const splitted = authHeader.split(' ');
    const token = splitted[1];
    validateBearerToken(token);
}

export function validateBearerToken(token) {
    if (token === JWT_BEARER_TOKEN) {
        return true;
    } else {
        console.log('token not valid ' + token);
        throw new Error('not authenticated');
    }
}

export async function run() {
    let documents = [];
    const app = express();
    app.use(cors());

    /**
     * In this example we generate the GraphQL schema from the RxDB schema.
     * Of course you could also write it by hand or extend and existing one.
     */
    const generatedSchema = graphQLSchemaFromRxSchema(graphQLGenerationInput);
    const graphQLSchema = generatedSchema.asString;

    console.log('Server side GraphQL Schema:');
    console.log(graphQLSchema);
    const schema = buildSchema(graphQLSchema);

    const pubsub = new PubSub();

    // The root provides a resolver function for each API endpoint
    const root = {
        pullHero: (args, request) => {
            log('## pullHero()');
            log(args);
            authenticateRequest(request);


            const lastId = args.checkpoint ? args.checkpoint.id : '';
            const minUpdatedAt = args.checkpoint ? args.checkpoint.updatedAt : 0;

            // sorted by updatedAt and primary
            const sortedDocuments = documents.sort(sortByUpdatedAtAndPrimary);

            // only return where updatedAt >= minUpdatedAt
            const filterForMinUpdatedAtAndId = sortedDocuments.filter(doc => {
                if (!args.checkpoint) {
                    return true;
                }
                if (doc.updatedAt < minUpdatedAt) {
                    return false;
                }
                if (doc.updatedAt > minUpdatedAt) {
                    return true;
                }
                if (doc.updatedAt === minUpdatedAt) {
                    if (doc.id > lastId) {
                        return true;
                    } else {
                        return false;
                    }
                }
            });

            // apply limit
            const limitedDocs = filterForMinUpdatedAtAndId.slice(0, args.limit);

            const last = lastOfArray(limitedDocs);
            const ret = {
                documents: limitedDocs,
                checkpoint: last ? {
                    id: last.id,
                    updatedAt: last.updatedAt
                } : {
                    id: lastId,
                    updatedAt: minUpdatedAt
                }
            };
            console.log('pullHero() ret:');
            console.log(JSON.stringify(ret, null, 4));
            return ret;
        },
        pushHero: (args, request) => {
            log('## pushHero()');
            log(args);
            authenticateRequest(request);

            const rows = args.heroPushRow;
            let lastCheckpoint = {
                id: '',
                updatedAt: 0
            };

            const conflicts = [];


            const writtenDocs = rows.map(row => {
                const docId = row.newDocumentState.id;
                const docCurrentMaster = documents.find(d => d.id === docId);

                /**
                 * Detect conflicts.
                 */
                if (
                    docCurrentMaster &&
                    row.assumedMasterState &&
                    docCurrentMaster.updatedAt !== row.assumedMasterState.updatedAt
                ) {
                    conflicts.push(docCurrentMaster);
                    return;
                }

                const doc = row.newDocumentState;
                documents = documents.filter(d => d.id !== doc.id);
                documents.push(doc);

                lastCheckpoint.id = doc.id;
                lastCheckpoint.updatedAt = doc.updatedAt;
                return doc;
            });

            pubsub.publish(
                'streamHero',
                {
                    streamHero: {
                        documents: writtenDocs,
                        checkpoint: lastCheckpoint
                    }
                }
            );


            console.log('## current documents:');
            console.log(JSON.stringify(documents, null, 4));<commit_msg>ADD conflict handling to GraphQL example 21<commit_after>import express from 'express';
import * as path from 'path';
const { graphqlHTTP } = require('express-graphql');
const cors = require('cors');
import { PubSub } from 'graphql-subscriptions';
import {
    buildSchema,
    execute,
    subscribe
} from 'graphql';
import { SubscriptionServer } from 'subscriptions-transport-ws';
import { createServer } from 'http';

import {
    GRAPHQL_PORT,
    GRAPHQL_PATH,
    GRAPHQL_SUBSCRIPTION_PORT,
    GRAPHQL_SUBSCRIPTION_PATH,
    graphQLGenerationInput,
    JWT_BEARER_TOKEN
} from '../shared';

import {
    graphQLSchemaFromRxSchema
} from 'rxdb/plugins/replication-graphql';

import {
    lastOfArray
} from 'rxdb';

function log(msg) {
    const prefix = '# GraphQL Server: ';
    if (typeof msg === 'string') {
        console.log(prefix + msg);
    } else {
        console.log(prefix + JSON.stringify(msg, null, 2));
    }
}

function sortByUpdatedAtAndPrimary(a, b) {
    if (a.updatedAt > b.updatedAt) return 1;
    if (a.updatedAt < b.updatedAt) return -1;

    if (a.updatedAt === b.updatedAt) {
        if (a.id > b.id) return 1;
        if (a.id < b.id) return -1;
        else return 0;
    }
}


/**
 * Returns true if the request is authenticated
 * throws if not.
 * In a real world app you would parse and validate the bearer token.
 * @link https://graphql.org/graphql-js/authentication-and-express-middleware/
 */
export function authenticateRequest(request) {
    const authHeader = request.header('authorization');
    const splitted = authHeader.split(' ');
    const token = splitted[1];
    validateBearerToken(token);
}

export function validateBearerToken(token) {
    if (token === JWT_BEARER_TOKEN) {
        return true;
    } else {
        console.log('token not valid ' + token);
        throw new Error('not authenticated');
    }
}

export async function run() {
    let documents = [];
    const app = express();
    app.use(cors());

    /**
     * In this example we generate the GraphQL schema from the RxDB schema.
     * Of course you could also write it by hand or extend and existing one.
     */
    const generatedSchema = graphQLSchemaFromRxSchema(graphQLGenerationInput);
    const graphQLSchema = generatedSchema.asString;

    console.log('Server side GraphQL Schema:');
    console.log(graphQLSchema);
    const schema = buildSchema(graphQLSchema);

    const pubsub = new PubSub();

    // The root provides a resolver function for each API endpoint
    const root = {
        pullHero: (args, request) => {
            log('## pullHero()');
            log(args);
            authenticateRequest(request);


            const lastId = args.checkpoint ? args.checkpoint.id : '';
            const minUpdatedAt = args.checkpoint ? args.checkpoint.updatedAt : 0;

            // sorted by updatedAt and primary
            const sortedDocuments = documents.sort(sortByUpdatedAtAndPrimary);

            // only return where updatedAt >= minUpdatedAt
            const filterForMinUpdatedAtAndId = sortedDocuments.filter(doc => {
                if (!args.checkpoint) {
                    return true;
                }
                if (doc.updatedAt < minUpdatedAt) {
                    return false;
                }
                if (doc.updatedAt > minUpdatedAt) {
                    return true;
                }
                if (doc.updatedAt === minUpdatedAt) {
                    if (doc.id > lastId) {
                        return true;
                    } else {
                        return false;
                    }
                }
            });

            // apply limit
            const limitedDocs = filterForMinUpdatedAtAndId.slice(0, args.limit);

            const last = lastOfArray(limitedDocs);
            const ret = {
                documents: limitedDocs,
                checkpoint: last ? {
                    id: last.id,
                    updatedAt: last.updatedAt
                } : {
                    id: lastId,
                    updatedAt: minUpdatedAt
                }
            };
            console.log('pullHero() ret:');
            console.log(JSON.stringify(ret, null, 4));
            return ret;
        },
        pushHero: (args, request) => {
            log('## pushHero()');
            log(args);
            authenticateRequest(request);

            const rows = args.heroPushRow;
            let lastCheckpoint = {
                id: '',
                updatedAt: 0
            };

            const conflicts = [];


            const writtenDocs = [];
            rows.forEach(row => {
                const docId = row.newDocumentState.id;
                const docCurrentMaster = documents.find(d => d.id === docId);

                /**
                 * Detect conflicts.
                 */
                if (
                    docCurrentMaster &&
                    row.assumedMasterState &&
                    docCurrentMaster.updatedAt !== row.assumedMasterState.updatedAt
                ) {
                    conflicts.push(docCurrentMaster);
                    return;
                }

                const doc = row.newDocumentState;
                documents = documents.filter(d => d.id !== doc.id);
                documents.push(doc);

                lastCheckpoint.id = doc.id;
                lastCheckpoint.updatedAt = doc.updatedAt;
                writtenDocs.push(doc);
            });

            pubsub.publish(
                'streamHero',
                {
                    streamHero: {
                        documents: writtenDocs,
                        checkpoint: lastCheckpoint
                    }
                }
            );


            console.log('## current documents:');
            console.log(JSON.stringify(documents, null, 4));
{'fraction_non_alphanumeric': 0.10105697344676463, 'fraction_numerical': 0.002234252814299218, 'mean_word_length': 1.600089365504915, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2468530', 'n_tokens_mistral': 3178, 'n_tokens_neox': 2987, 'n_words': 740}