import { readFileSync } from 'fs';
import Redis from 'ioredis';
import Redlock from 'redlock';
import { RedisPubSub } from 'graphql-redis-subscriptions';
import * as R from 'ramda';
import { create as createJsonDiff } from 'jsondiffpatch';
import conf, { booleanConf, configureCA, DEV_MODE, ENABLED_CACHING, logApp } from '../config/conf';
import {
  generateCreateMessage,
  generateDeleteMessage,
  generateMergeMessage,
  generateUpdateMessage,
  isEmptyField,
  isInferredIndex,
  isSyncTesting,
  UPDATE_OPERATION_ADD,
  UPDATE_OPERATION_REMOVE,
} from './utils';
import { isStixObject } from '../schema/stixCoreObject';
import { isStixRelationship } from '../schema/stixRelationship';
import { EVENT_TYPE_CREATE, EVENT_TYPE_DELETE, EVENT_TYPE_MERGE, EVENT_TYPE_UPDATE } from './rabbitmq';
import { buildStixData, updateInputsToPatch } from './stix';
import { DatabaseError, FunctionalError, UnsupportedError } from '../config/errors';
import { now } from '../utils/format';
import RedisStore from './sessionStore-redis';
import SessionStoreMemory from './sessionStore-memory';
import {
  isStixMetaRelationship,
  RELATION_CREATED_BY,
  RELATION_OBJECT_MARKING,
  STIX_META_RELATION_TO_OPENCTI_INPUT,
} from '../schema/stixMetaRelationship';
import { isStixCyberObservableRelationship } from '../schema/stixCyberObservableRelationship';
import { getInstanceIdentifiers, getInstanceIds } from '../schema/identifier';
import { BASE_TYPE_RELATION } from '../schema/general';

const USE_SSL = booleanConf('redis:use_ssl', false);
const REDIS_CA = conf.get('redis:ca').map((path) => readFileSync(path));

const BASE_DATABASE = 0; // works key for tracking / stream
const CONTEXT_DATABASE = 1; // locks / user context
const OPENCTI_STREAM = 'stream.opencti';
const REDIS_EXPIRE_TIME = 90;

const redisOptions = (database) => ({
  db: database,
  port: conf.get('redis:port'),
  host: conf.get('redis:hostname'),
  username: conf.get('redis:username'),
  password: conf.get('redis:password'),
  tls: USE_SSL ? configureCA(REDIS_CA) : null,
  retryStrategy: /* istanbul ignore next */ (times) => Math.min(times * 50, 2000),
  lazyConnect: true,
  enableAutoPipelining: false,
  enableOfflineQueue: true,
  maxRetriesPerRequest: 30,
  showFriendlyErrorStack: DEV_MODE,
});

export const pubsub = new RedisPubSub({
  publisher: new Redis(redisOptions()),
  subscriber: new Redis(redisOptions()),
});
const createRedisClient = (provider, database = BASE_DATABASE) => {
  const client = new Redis(redisOptions(database));
  client.on('close', () => logApp.info(`[REDIS] Redis '${provider}' client closed`));
  client.on('ready', () => logApp.info(`[REDIS] Redis '${provider}' client ready`));
  client.on('reconnecting', () => logApp.info(`[REDIS] '${provider}' Redis client reconnecting`));
  client.defineCommand('cacheGet', {
    lua:
      'local index = 1\n' +
      "local resolvedKeys = redis.call('mget', unpack(KEYS))\n" +
      'for p, k in pairs(resolvedKeys) do \n' +
      '    if (k==nil or (type(k) == "boolean" and not k)) then \n' +
      '        index = index+1\n' +
      '    elseif (k:sub(0, 1) == "@") then \n' +
      '        local subKey = "cache:" .. k:sub(2, #k)\n' +
      "        resolvedKeys[index] = redis.call('get', subKey)\n" +
      '        index = index+1\n' +
      '    else \n' +
      '        index = index+1\n' +
      '    end\n' +
      'end\n' +
      'return resolvedKeys\n',
  });
  return client;
};
const clientBase = createRedisClient('Client base');
const clientCache = createRedisClient('Client cache');
const clientContext = createRedisClient('Client context', CONTEXT_DATABASE);

export const createMemorySessionStore = () => {
  return new SessionStoreMemory({
    checkPeriod: 3600000, // prune expired entries every 1h
  });
};
export const createRedisSessionStore = () => {
  return new RedisStore(clientContext);
};

export const redisIsAlive = async () => {
  try {
    await clientBase.get('test-key');
  } catch {
    throw DatabaseError('Redis seems down');
  }
  return true;
};
export const getRedisVersion = async () => {
  await clientBase.call('INFO');
  return clientBase.serverInfo.redis_version;
};

/* istanbul ignore next */
export const notify = (topic, instance, user, context) => {
  pubsub.publish(topic, { instance, user, context });
  return instance;
};

// region user context (clientContext)
const contextFetchMatch = async (match) => {
  return new Promise((resolve, reject) => {
    const elementsPromise = [];
    const stream = clientContext.scanStream({
      match,
      count: 100,
    });
    stream.on('data', (resultKeys) => {
      for (let i = 0; i < resultKeys.length; i += 1) {
        const resultKey = resultKeys[i];
        elementsPromise.push(clientContext.get(resultKey).then((d) => ({ key: resultKey, value: d })));
      }
    });
    stream.on('error', (error) => {
      /* istanbul ignore next */
      reject(error);
    });
    stream.on('end', () => {
      Promise.all(elementsPromise).then((data) => {
        const elements = R.map((d) => ({ redis_key: d.key, ...JSON.parse(d.value) }), data);
        resolve(elements);
      });
    });
  });
};
export const setEditContext = async (user, instanceId, input) => {
  const data = R.assoc('name', user.user_email, input);
  return clientContext.set(
    `edit:${instanceId}:${user.id}`,
    JSON.stringify(data),
    'ex',
    5 * 60 // Key will be remove if user is not active during 5 minutes
  );
};
export const fetchEditContext = async (instanceId) => {
  return contextFetchMatch(`edit:${instanceId}:*`);
};
export const delEditContext = async (user, instanceId) => {
  return clientContext.del(`edit:${instanceId}:${user.id}`);
};
export const delUserContext = async (user) => {
  return new Promise((resolve, reject) => {
    const stream = clientContext.scanStream({
      match: `*:*:${user.id}`,
      count: 100,
    });
    const keys = [];
    stream.on('data', (resultKeys) => {
      for (let index = 0; index < resultKeys.length; index += 1) {
        keys.push(resultKeys[index]);
      }
    });
    stream.on('error', (error) => {
      /* istanbul ignore next */
      reject(error);
    });
    stream.on('end', () => {
      if (!R.isEmpty(keys)) {
        clientContext.del(keys);
      }
      resolve();
    });
  });
};
// endregion

// region basic operations
export const redisTx = async (client, callback) => {
  const tx = client.multi();
  try {
    await callback(tx);
    return tx.exec();
  } catch (e) {
    throw DatabaseError('Redis Tx error', { error: e });
  }
};
export const updateObjectRaw = async (tx, id, input) => {
  const data = R.flatten(R.toPairs(input));
  await tx.call('HSET', id, data);
};
export const updateObjectCounterRaw = async (tx, id, field, number) => {
  await tx.call('HINCRBY', id, field, number);
};
// endregion

// region concurrent deletion
export const redisAddDeletions = async (internalIds) => {
  const deletionId = new Date().getTime();
  const ids = Array.isArray(internalIds) ? internalIds : [internalIds];
  return redisTx(clientContext, (tx) => {
    tx.call('SETEX', `deletion-${deletionId}`, REDIS_EXPIRE_TIME, JSON.stringify(ids));
  });
};
export const redisFetchLatestDeletions = async () => {
  const keys = await contextFetchMatch('deletion-*');
  return R.uniq(R.flatten(keys));
};
// endregion

// region locking (clientContext)
const checkParticipantsDeletion = async (participantIds) => {
  const latestDeletions = await redisFetchLatestDeletions();
  const deletedParticipantsIds = participantIds.filter((x) => latestDeletions.includes(x));
  if (deletedParticipantsIds.length > 0) {
    // noinspection ExceptionCaughtLocallyJS
    throw FunctionalError('Cant update an element based on deleted dependencies', { deletedParticipantsIds });
  }
};
export const lockResource = async (resources, automaticExtension = true) => {
  let timeout;
  const locks = R.uniq(resources);
  const automaticExtensionThreshold = conf.get('app:concurrency:extension_threshold');
  const retryCount = conf.get('app:concurrency:retry_count');
  const retryDelay = conf.get('app:concurrency:retry_delay');
  const retryJitter = conf.get('app:concurrency:retry_jitter');
  const maxTtl = conf.get('app:concurrency:max_ttl');
  const redlock = new Redlock([clientContext], { retryCount, retryDelay, retryJitter });
  // Get the lock
  const lock = await redlock.lock(locks, maxTtl); // Force unlock after maxTtl
  let expiration = Date.now() + maxTtl;
  const extend = async () => {
    try {
      await lock.extend(maxTtl);
      expiration = Date.now() + maxTtl;
      if (automaticExtension) {
        // eslint-disable-next-line no-use-before-define
        queue();
      }
    } catch (e) {
      logApp.debug('[REDIS] Failed to extend resource', { locks });
    }
  };
  const queue = () => {
    const timeToWait = expiration - Date.now() - automaticExtensionThreshold;
    timeout = setTimeout(() => extend(), timeToWait);
  };
  if (automaticExtension) {
    queue();
  }
  // If lock succeed we need to be sure that delete occurred just before the resolution/lock
  await checkParticipantsDeletion(resources);
  // Return the lock and capable actions
  return {
    extend,
    unlock: async () => {
      // First clear the auto extends if needed
      if (timeout) {
        clearTimeout(timeout);
        timeout = undefined;
      }
      // Then unlock in redis
      try {
        await lock.unlock();
      } catch (e) {
        logApp.debug('[REDIS] Failed to unlock resource', { locks });
      }
    },
  };
};
// endregion

// region cache
const cacheExtraIds = (e) => getInstanceIds(e, true).map((i) => `cache:${i}`);
export const cacheSet = async (elements) => {
  if (ENABLED_CACHING) {
    await redisTx(clientCache, (tx) => {
      for (let index = 0; index < elements.length; index += 1) {
        const element = elements[index];
        tx.call('SET', `cache:${element.internal_id}`, JSON.stringify(element), 'ex', 5 * 60);
        const ids = cacheExtraIds(element);
        for (let indexId = 0; indexId < ids.length; indexId += 1) {
          const id = ids[indexId];
          tx.call('SET', id, `@${element.internal_id}`, 'ex', 5 * 60);
        }
      }
    });
  }
};
export const cacheDel = async (elements) => {
  if (ENABLED_CACHING) {
    const ids = R.flatten(elements.map((e) => [`cache:${e.internal_id}`, ...cacheExtraIds(e)]));
    await clientCache.del(ids);
  }
};
export const cachePurge = async () => {
  if (ENABLED_CACHING) {
    const keys = await clientCache.keys('cache:*');
    if (keys && keys.length > 0) {
      await clientCache.del(keys);
    }
  }
};
export const cacheGet = async (id) => {
  const ids = Array.isArray(id) ? id : [id];
  if (ENABLED_CACHING) {
    const result = {};
    if (ids.length > 0) {
      // eslint-disable-next-line prettier/prettier
      const keyValues = await clientCache.cacheGet(ids.length, ids.map((i) => `cache:${i}`));
      for (let index = 0; index < ids.length; index += 1) {
        const val = keyValues[index];
        result[ids[index]] = val ? JSON.parse(val) : val;
      }
    }
    return result;
  }
  return undefined;
};
// endregion

// region opencti stream
const streamTrimming = conf.get('redis:trimming') || 0;
const mapJSToStream = (event) => {
  const cmdArgs = [];
  Object.keys(event).forEach((key) => {
    const value = event[key];
    if (value !== undefined) {
      cmdArgs.push(key);
      cmdArgs.push(JSON.stringify(value));
    }
  });
  return cmdArgs;
};
export const buildEvent = (eventType, user, markings, message, data) => {
  if (!data.id || !data.x_opencti_id || !data.type) {
    throw UnsupportedError('Stream event requires id, type and x_opencti_id');
  }
  return {
    version: '3', // Event version.
    type: eventType,
    origin: user.origin,
    markings: markings || [],
    message,
    data,
  };
};
const pushToStream = (client, event) => {
  // Event can be empty because of UUIDv1 in STIX IDs
  if (!event) {
    return true;
  }
  if (streamTrimming) {
    return client.call('XADD', OPENCTI_STREAM, 'MAXLEN', '~', streamTrimming, '*', ...mapJSToStream(event));
  }
  return client.call('XADD', OPENCTI_STREAM, '*', ...mapJSToStream(event));
};
const DIFF_ADDED = 1;
const DIFF_CHANGE = 2;
const DIFF_REMOVE = 3;
const DIFF_TYPE = '_t';
const DIFF_TYPE_ARRAY = 'a';
export const computeMergeDifferential = (initialInstance, mergedInstance) => {
  const convertInit = buildStixData(initialInstance, { patchGeneration: true });
  const convertMerged = buildStixData(mergedInstance, { patchGeneration: true });
  const diffGenerator = createJsonDiff({
    objectHash: (obj) => {
      return obj.x_opencti_internal_id;
    },
  });
  const diff = diffGenerator.diff(convertInit, convertMerged);
  const patch = {};
  const entries = Object.entries(diff);
  for (let index = 0; index < entries.length; index += 1) {
    const [field, diffDelta] = entries[index];
    // https://github.com/benjamine/jsondiffpatch/blob/master/docs/deltas.md
    if (Array.isArray(diffDelta)) {
      let current;
      let previous;
      // Value added
      if (diffDelta.length === DIFF_ADDED) {
        const value = R.head(diffDelta);
        current = value;
        previous = Array.isArray(value) ? [] : '';
      }
      // Value changed
      if (diffDelta.length === DIFF_CHANGE) {
        current = R.last(diffDelta);
        previous = R.head(diffDelta);
      }
      // Value removed
      if (diffDelta.length === DIFF_REMOVE) {
        const value = R.head(diffDelta);
        previous = value;
        current = Array.isArray(value) ? [] : '';
      }
      // Setup the patch
      if (patch.replace) {
        patch.replace[field] = { current, previous };
      } else {
        patch.replace = { [field]: { current, previous } };
      }
    } else if (diffDelta[DIFF_TYPE] === DIFF_TYPE_ARRAY) {
      // Is an array changes
      const delta = R.dissoc(DIFF_TYPE, diffDelta);
      const deltaObjEntries = Object.entries(delta);
      for (let indexDelta = 0; indexDelta < deltaObjEntries.length; indexDelta += 1) {
        const [, diffData] = deltaObjEntries[indexDelta];
        if (diffData.length === DIFF_ADDED) {
          if (patch.add) {
            patch.add[field] = diffData;
          } else {
            patch.add = { [field]: diffData };
          }
        }
        if (diffData.length === DIFF_REMOVE) {
          const removedValue = R.head(diffData);
          const removeVal = Array.isArray(removedValue) ? removedValue : [removedValue];
          if (patch.remove) {
            patch.remove[field] = removeVal;
          } else {
            patch.remove = { [field]: removeVal };
          }
        }
      }
    } else {
      // Is a internal complex object, like extensions
      // TODO @JRI
    }
  }
  return patch;
};
// Merge
const buildMergeEvent = (user, initialInstance, mergedInstance, sourceEntities, impacts) => {
  const patch = computeMergeDifferential(initialInstance, mergedInstance);
  const message = generateMergeMessage(initialInstance, sourceEntities);
  const data = buildStixData(mergedInstance, { clearEmptyValues: true });
  const { updatedRelations, dependencyDeletions } = impacts;
  data.x_opencti_patch = patch;
  data.x_opencti_context = {
    sources: R.map((s) => buildStixData(s, { clearEmptyValues: true }), sourceEntities),
    deletions: R.map((s) => buildStixData(s, { clearEmptyValues: true }), dependencyDeletions),
    shifts: updatedRelations,
  };
  return buildEvent(EVENT_TYPE_MERGE, user, mergedInstance.object_marking_refs, message, data);
};
export const storeMergeEvent = async (user, initialInstance, mergedInstance, sourceEntities, impacts) => {
  try {
    const event = buildMergeEvent(user, initialInstance, mergedInstance, sourceEntities, impacts);
    // Push the event in the stream only if instance is in "real index"
    if (!isInferredIndex(mergedInstance._index) && !isSyncTesting(user)) {
      await pushToStream(clientBase, event);
    }
  } catch (e) {
    throw DatabaseError('Error in store merge event', { error: e });
  }
};
// Update
export const buildUpdateEvent = (user, instance, patch, opts = {}) => {
  const { withoutMessage = false, clearEmptyValues = false } = opts;
  // dataUpdate can be empty
  if (isEmptyField(patch)) {
    return null;
  }
  // Build the final data
  const data = { ...instance, x_opencti_patch: patch };
  // Generate the message
  const message = withoutMessage ? '-' : generateUpdateMessage(patch);
  // Build and send the event
  const dataEvent = buildStixData(data, { clearEmptyValues });
  return buildEvent(EVENT_TYPE_UPDATE, user, instance.object_marking_refs, message, dataEvent);
};
export const storeUpdateEvent = async (user, instance, patchInputs, opts = {}) => {
  const { mustBeRepublished = false } = opts;
  // updateEvents -> [{ operation, input }]
  if (isStixObject(instance.entity_type) || isStixRelationship(instance.entity_type)) {
    try {
      const patch = updateInputsToPatch(patchInputs);
      const eventData = mustBeRepublished ? instance : getInstanceIdentifiers(instance);
      const event = buildUpdateEvent(user, eventData, patch, opts);
      // Push the event in the stream only if instance is in "real index"
      if (!isInferredIndex(instance._index) && !isSyncTesting(user)) {
        await pushToStream(clientBase, event);
      }
      return event;
    } catch (e) {
      throw DatabaseError('Error in store update event', { error: e });
    }
  }
  return null;
};
// Create
export const buildCreateEvent = async (user, instance, input, loaders, opts = {}) => {
  const { withoutMessage = false } = opts;
  const { stixLoadById, connectionLoaders } = loaders;
  // If internal relation, publish an update instead of a creation
  if (isStixCyberObservableRelationship(instance.entity_type) || isStixMetaRelationship(instance.entity_type)) {
    const mustRepublished = instance.entity_type === RELATION_OBJECT_MARKING;
    let publishedInstance;
    if (mustRepublished) {
      publishedInstance = await stixLoadById(user, instance.from.internal_id);
    } else if (instance.from.base_type === BASE_TYPE_RELATION) {
      const instanceWithConnections = await connectionLoaders(user, instance.from);
      publishedInstance = getInstanceIdentifiers(instanceWithConnections);
    } else {
      publishedInstance = getInstanceIdentifiers(instance.from);
    }
    const key = STIX_META_RELATION_TO_OPENCTI_INPUT[instance.entity_type];
    if (instance.entity_type === RELATION_CREATED_BY) {
      // eslint-disable-next-line prettier/prettier
      const inputVal = { key, value: [instance.to], previous: null };
      const patch = updateInputsToPatch([inputVal]);
      return buildUpdateEvent(user, publishedInstance, patch, opts);
    }
    const inputVal = { key, value: [instance.to], operation: UPDATE_OPERATION_ADD };
    const patch = updateInputsToPatch([inputVal]);
    return buildUpdateEvent(user, publishedInstance, patch, opts);
  }
  // Convert the input to data
  const mergedData = { ...instance, ...input };
  const data = buildStixData(mergedData, { clearEmptyValues: true });
  // Generate the message
  const message = withoutMessage ? '-' : generateCreateMessage(mergedData);
  // Build and send the event
  const inputMarkings = (input.objectMarking || []).map((m) => m.internal_id);
  return buildEvent(EVENT_TYPE_CREATE, user, inputMarkings, message, data);
};
export const buildScanEvent = (user, instance) => {
  const data = buildStixData(instance);
  return buildEvent(EVENT_TYPE_CREATE, user, instance.object_marking_refs ?? [], '-', data);
};
export const storeCreateEvent = async (user, instance, input, loaders) => {
  if (isStixObject(instance.entity_type) || isStixRelationship(instance.entity_type)) {
    try {
      const event = await buildCreateEvent(user, instance, input, loaders);
      // Push the event in the stream only if instance is in "real index"
      if (!isInferredIndex(instance._index) && !isSyncTesting(user)) {
        await pushToStream(clientBase, event);
      }
      return event;
    } catch (e) {
      throw DatabaseError('Error in store create event', { error: e });
    }
  }
  return null;
};
// Delete
export const buildDeleteEvent = async (user, instance, dependencyDeletions, loaders, opts = {}) => {
  const { withoutMessage = false } = opts;
  const { stixLoadById, connectionLoaders } = loaders;
  // If internal relation, publish an update instead of a creation
  if (isStixCyberObservableRelationship(instance.entity_type) || isStixMetaRelationship(instance.entity_type)) {
    const mustRepublished = instance.entity_type === RELATION_OBJECT_MARKING;
    let publishedInstance;
    if (mustRepublished) {
      publishedInstance = await stixLoadById(user, instance.from.internal_id);
    } else if (instance.from.base_type === BASE_TYPE_RELATION) {
      const instanceWithConnections = await connectionLoaders(user, instance.from);
      publishedInstance = getInstanceIdentifiers(instanceWithConnections);
    } else {
      publishedInstance = getInstanceIdentifiers(instance.from);
    }
    const key = STIX_META_RELATION_TO_OPENCTI_INPUT[instance.entity_type];
    if (instance.entity_type === RELATION_CREATED_BY) {
      const inputVal = { key, value: null, previous: [instance.to] };
      const patch = updateInputsToPatch([inputVal]);
      return buildUpdateEvent(user, publishedInstance, patch, opts);
    }
    const inputVal = { key, value: [instance.to], operation: UPDATE_OPERATION_REMOVE };
    const patch = updateInputsToPatch([inputVal]);
    return buildUpdateEvent(user, publishedInstance, patch, opts);
  }
  // Convert the input to data
  const data = buildStixData(instance, { clearEmptyValues: true });
  // Generate the message
  const message = withoutMessage ? '-' : generateDeleteMessage(instance);
  data.x_opencti_context = {
    deletions: R.map((s) => buildStixData(s, { clearEmptyValues: true }), dependencyDeletions),
  };
  return buildEvent(EVENT_TYPE_DELETE, user, instance.object_marking_refs, message, data);
};
export const storeDeleteEvent = async (user, instance, dependencyDeletions, loaders) => {
  try {
    if (isStixObject(instance.entity_type) || isStixRelationship(instance.entity_type)) {
      const event = await buildDeleteEvent(user, instance, dependencyDeletions, loaders);
      // Push the event in the stream only if instance is in "real index"
      if (!isInferredIndex(instance._index) && !isSyncTesting(user)) {
        await pushToStream(clientBase, event);
      }
      return event;
    }
  } catch (e) {
    throw DatabaseError('Error in store delete event', { error: e });
  }
  return null;
};

const fetchStreamInfo = async () => {
  const res = await clientBase.call('XINFO', 'STREAM', OPENCTI_STREAM);
  const [, size, , , , , , lastId, , , , , ,] = res;
  return { lastEventId: lastId, streamSize: size };
};
const mapStreamToJS = ([id, data]) => {
  const count = data.length / 2;
  const result = { eventId: id };
  for (let i = 0; i < count; i += 1) {
    result[data[2 * i]] = JSON.parse(data[2 * i + 1]);
  }
  return result;
};
const processStreamResult = async (results, callback) => {
  const streamData = R.map((r) => mapStreamToJS(r), results);
  const lastElement = R.last(streamData);
  // Prepare the elements
  const processedResults = [];
  for (let index = 0; index < streamData.length; index += 1) {
    const dataElement = streamData[index];
    const { eventId, type, markings, origin, data, message, version } = dataElement;
    const eventData = { markings, origin, data, message, version };
    processedResults.push({ id: eventId, topic: type, data: eventData });
  }
  // Callback the data
  await callback(processedResults);
  return lastElement.eventId;
};

let processingLoopPromise;
const WAIT_TIME = 1000;
const MAX_RANGE_MESSAGES = 500;
export const createStreamProcessor = (user, provider, callback, maxRange = MAX_RANGE_MESSAGES) => {
  let client;
  let startEventId;
  let streamListening = true;
  const processInfo = async () => {
    return fetchStreamInfo();
  };
  const processStep = async () => {
    // since previous call is async (and blocking) we should check if we are still running before processing the message
    if (!streamListening) {
      return false;
    }
    const streamResult = await client.xread(
      'BLOCK',
      WAIT_TIME,
      'COUNT',
      maxRange,
      'STREAMS',
      OPENCTI_STREAM,
      startEventId
    );
    if (streamResult && streamResult.length > 0) {
      const [, results] = R.head(streamResult);
      const lastElementId = await processStreamResult(results, callback);
      startEventId = lastElementId || startEventId;
    }
    return true;
  };
  const processingLoop = async () => {
    while (streamListening) {
      if (!(await processStep())) {
        break;
      }
    }
  };
  return {
    info: async () => processInfo(),
    start: async (start = 'live') => {
      let fromStart = start;
      if (isEmptyField(fromStart)) {
        fromStart = 'live';
      }
      startEventId = fromStart === 'live' ? '$' : fromStart;
      client = await createRedisClient(provider); // Create client for this processing loop
      logApp.info(`[STREAM] Starting stream processor for ${provider}`);
      processingLoopPromise = processingLoop();
    },
    shutdown: async () => {
      logApp.info(`[STREAM] Shutdown stream processor for ${user.user_email}`);
      streamListening = false;
      if (processingLoopPromise) {
        await processingLoopPromise;
      }
      if (client) {
        await client.disconnect();
      }
    },
  };
};
// endregion

// region work handling
export const redisDeleteWork = async (internalIds) => {
  const ids = Array.isArray(internalIds) ? internalIds : [internalIds];
  return redisTx(clientBase, (tx) => {
    tx.call('DEL', ...ids);
  });
};
export const redisCreateWork = async (element) => {
  return redisTx(clientBase, (tx) => {
    const data = R.flatten(R.toPairs(element));
    tx.call('HSET', element.internal_id, data);
  });
};
export const redisGetWork = async (internalId) => {
  const rawElement = await clientBase.call('HGETALL', internalId);
  return R.fromPairs(R.splitEvery(2, rawElement));
};
export const redisUpdateWork = async (id, input) => {
  const data = R.flatten(R.toPairs(input));
  return redisTx(clientBase, (tx) => {
    tx.call('HSET', id, data);
  });
};
export const redisUpdateWorkFigures = async (workId) => {
  const timestamp = now();
  const [, , fetched] = await redisTx(clientBase, async (tx) => {
    await updateObjectCounterRaw(tx, workId, 'import_processed_number', 1);
    await updateObjectRaw(tx, workId, { import_last_processed: timestamp });
    await tx.call('HGETALL', workId);
  });
  const updatedMetrics = R.fromPairs(R.splitEvery(2, R.last(fetched)));
  const { import_processed_number: pn, import_expected_number: en } = updatedMetrics;
  return { isComplete: parseInt(pn, 10) === parseInt(en, 10), total: pn, expected: en };
};
export const redisUpdateActionExpectation = async (user, workId, expectation) => {
  await redisTx(clientBase, async (tx) => {
    await updateObjectCounterRaw(tx, workId, 'import_expected_number', expectation);
  });
  return workId;
};
// endregion
