index.ts

Filename	Percent	Threshold	Total	Covered	Uncovered
src/results/index.ts	98.85%	80%	1219	1205	14

import {
  COUNTY_VOTE,
  DELEGATE_ALLOCATION,
  DELEGATE_VOTE,
  type DocType,
  EDITORIAL_TURNOUT,
  ELECTORAL_VOTE,
  EST,
  MSG,
  NATIONAL_DELEGATE_SUMMARY,
  TOTAL_EXPECTED_VOTE,
} from 'constants/docTypes.js';
import { type DB, Query } from 'db/index.js';
import { type GenericDocument, uniqByKeyandMaxSeq } from 'utils/uniqs.js';
import { type GenericDocumentKey, parseKey } from 'utils/documentKey.js';
import { and, eq, inArray, isNull } from 'drizzle-orm';

import { DBTransactionError } from '../exceptions/errors.js';
import { Endpoints } from '../endpoints/index.js';
import type { Logger } from 'winston';
import { call } from 'db/schema/call.js';
import { candidateElectoralVote } from 'db/schema/candidateElectoralVote.js';
import { candidateRace } from 'db/schema/candidateRace.js';
import { candidateRace_office } from 'db/schema/candidateRace_office.js';
import { candidateVote } from 'db/schema/candidateVote.js';
import { candidateVoteByParty } from 'db/schema/candidateVoteByParty.js';
import countyVoteParsers from './parsers/county-vote/index.js';
import { delegateAllocation } from 'db/schema/delegateAllocation.js';
import delegateAllocationParsers from './parsers/delegate-allocation/index.js';
import delegateSummaryParsers from './parsers/national-delegate-summary/index.js';
import delegateVoteParsers from './parsers/delegate-vote/index.js';
import { documentSequence } from 'db/schema/documentSequence.js';
import { editorialTurnout } from 'db/schema/editorialTurnout.js';
import editorialTurnoutParsers from './parsers/editorial-turnout/index.js';
import { electoralVote } from 'db/schema/electoralVote.js';
import electoralVoteParsers from './parsers/electoral-vote/index.js';
import estParsers from './parsers/est/index.js';
import { jurisdictionRace } from 'db/schema/jurisdictionRace.js';
import { message } from 'db/schema/message.js';
import msgParsers from './parsers/msg/index.js';
import { nationalDelegateSummary } from 'db/schema/nationalDelegateSummary.js';
import { officeVote } from 'db/schema/officeVote.js';
import { presidentialPrimaryDistrictVoteByCounty } from 'db/schema/presidentialPrimaryDistrictVoteByCounty.js';
import { presidentialPrimaryVoteByDistrict } from 'db/schema/presidentialPrimaryVoteByDistrict.js';
import { stateDelegateSummary } from 'db/schema/stateDelegateSummary.js';
import { streamPosition } from 'db/schema/streamPosition.js';
import streamsParsers from './parsers/streams/index.js';
import { totalExpectedVote } from 'db/schema/totalExpectedVote.js';
import totalExpectedVoteParsers from './parsers/total-expected-vote/index.js';
import { voteFlag } from 'db/schema/voteFlag.js';

type Config = {
  apiKey: string;
  eventId: string;
  db: DB;
  logger?: Logger;
};

export class Results extends Endpoints {
  query: Query;
  constructor({ logger, apiKey, eventId, db }: Config) {
    super({ logger, apiKey, eventId, db });
    this.query = new Query(db, logger);
  }

  /**
   * Get the latest stream position for a particular doc type with filters.
   *
   * @param docTypesFilters Array of doc type constants
   * @return Latest positions
   */
  async getStreamPositions(docTypesFilters: string[] = []) {
    let positions = await this.getStreams();
    if (docTypesFilters)
      positions = positions.filter(({ docType }) =>
        docTypesFilters.includes(docType)
      );

    const streamPositions: (typeof streamPosition.$inferInsert)[] = [];

    for (const position of positions) {
      const { docType, maxPosition } = position;
      const { st: filterSt, ofc: filterOfc, scp: filterScp } = this.filter();
      const states = filterSt ? filterSt.split(',').sort() : [];
      if (states.length === 0) {
        streamPositions.push({
          uid: streamsParsers.parseStreamPositionId(
            docType,
            filterScp,
            filterOfc
          ),
          docType,
          filterScp,
          filterOfc,
          maxPosition,
        });
      }
      for (const filterSt of states) {
        streamPositions.push({
          uid: streamsParsers.parseStreamPositionId(
            docType,
            filterScp,
            filterOfc,
            filterSt
          ),
          docType,
          filterScp,
          filterOfc,
          filterSt,
          maxPosition,
        });
      }
    }
    return streamPositions;
  }

  /**
   * Update StreamPosition model for a particular doc type with filters.
   * @param docType doc type constant
   * @param newPosition new stream position
   */
  async updateStreamPosition(docType: DocType, newPosition: number) {
    const filter = this.filter();
    const { st: filterSt, ofc: filterOfc, scp: filterScp } = filter;
    const states = filterSt ? filterSt.split(',').sort() : [];

    if (states.length === 0) {
      const uid = streamsParsers.parseStreamPositionId(
        docType,
        filterScp,
        filterOfc
      );
      const streamPositionRecord = {
        docType,
        filterScp,
        filterOfc,
        maxPosition: newPosition,
      };
      await this.db
        .insert(streamPosition)
        .values({
          uid,
          ...streamPositionRecord,
        })
        .onConflictDoUpdate({
          target: streamPosition.uid,
          set: streamPositionRecord,
        });
    } else {
      for (const filterSt of states) {
        const uid = streamsParsers.parseStreamPositionId(
          docType,
          filterScp,
          filterOfc,
          filterSt
        );
        const streamPositionRecord = {
          docType,
          filterScp,
          filterOfc,
          filterSt,
          maxPosition: newPosition,
        };
        await this.db
          .insert(streamPosition)
          .values({
            uid,
            ...streamPositionRecord,
          })
          .onConflictDoUpdate({
            target: streamPosition.uid,
            set: streamPositionRecord,
          });
      }
    }
    await this.updateSupersededStreamPositions(docType, newPosition, filter);
  }

  /**
   * Update filtered stream positions that are superseded by unfiltered stream positions.
   *
   * When a stream position for a particular doc type is updated without a filter for office
   * or states, it should update stream positions that were previously set with a filter, since
   * those states and offices are included in the documents retrieved without a filter.
   *
   * @param docType doc type constant
   * @param newPosition new stream position
   * @param filter current filter set
   */
  async updateSupersededStreamPositions(
    docType: string,
    newPosition: number,
    filter: ReturnType<Results['filter']>
  ) {
    const { st: filterSt, ofc: filterOfc, scp: filterScp } = filter;

    if (!filterOfc) {
      await this.db
        .update(streamPosition)
        .set({ maxPosition: newPosition })
        .where(
          and(
            eq(streamPosition.docType, docType),
            eq(streamPosition.filterScp, filterScp),
            filterSt
              ? inArray(streamPosition.filterSt, filterSt.split(','))
              : isNull(streamPosition.filterSt)
          )
        );
    }

    if (!filterSt) {
      await this.db
        .update(streamPosition)
        .set({ maxPosition: newPosition })
        .where(
          and(
            eq(streamPosition.docType, docType),
            eq(streamPosition.filterScp, filterScp),
            filterOfc
              ? eq(streamPosition.filterOfc, filterOfc)
              : isNull(streamPosition.filterOfc)
          )
        );
    }
  }

  async updateSupersededStreamPositionsFromSweep(positions) {
    const filter = this.filter();
    const { ofc, st } = filter;
    // If both superseding fields are filtered, we can skip this step...
    if (ofc && st) return;
    for (const position of positions) {
      const { docType, maxPosition } = position;
      await this.updateSupersededStreamPositions(docType, maxPosition, filter);
    }
  }

  /**
   * Get the latest document sequence numbers.
   */
  async getDocumentSequences(documentSequenceUids: string[], docType: string) {
    return this.db
      .select({
        uid: documentSequence.uid,
        sequence: documentSequence.sequence,
      })
      .from(documentSequence)
      .where(
        and(
          inArray(documentSequence.uid, documentSequenceUids),
          eq(documentSequence.docType, docType)
        )
      );
  }

  /**
   * Filters documents using the document sequence number and returns
   * any documents newer than those already in the database.
   *
   * @param documents
   * @param docType Doc type constant
   * @param keyParser A function that takes a document and returns a unique key for that document
   */
  async getNewDocuments(
    documents: GenericDocument[],
    docType: string,
    keyParser: (doc: unknown) => string
  ) {
    const documentSequenceUids = documents.map((doc) => keyParser(doc));

    const sequences = await this.getDocumentSequences(
      documentSequenceUids,
      docType
    );

    return uniqByKeyandMaxSeq(documents).filter((doc) => {
      const documentSequnceUid = keyParser(doc);
      const { seq: currentSequence } = doc;
      const previousSequence = sequences.find(
        (s) => s.uid === documentSequnceUid
      )?.sequence;
      // The document has never been saved
      if (!previousSequence) return true;
      return currentSequence > previousSequence;
    });
  }

  private async makeCountyVotes(votes) {
    const candidateVotes = countyVoteParsers.parseCandidateVotes(votes);
    const candidateVoteByParties =
      countyVoteParsers.parseCandidateVoteByParties(votes);
    const voteFlags = countyVoteParsers.parseVoteFlags(votes);
    const officeVotes = countyVoteParsers.parseOfficeVotes(votes);
    const documentSequences = countyVoteParsers.parseDocumentSequences(votes);

    await this.query.bulkUpsert(candidateVote, candidateVotes);
    await this.query.bulkUpsert(candidateVoteByParty, candidateVoteByParties);
    await this.query.bulkUpsert(voteFlag, voteFlags);
    await this.query.bulkUpsert(officeVote, officeVotes);
    await this.query.bulkUpsert(documentSequence, documentSequences);
  }

  async updateCountyVotes() {
    const positions = await this.getStreamPositions([COUNTY_VOTE]);
    const votes = await this.getResultsCountyVote();

    await this.makeCountyVotes(votes);
    await this.query.bulkUpsert(streamPosition, positions);
    await this.updateSupersededStreamPositionsFromSweep(positions);
  }

  async updateStreamedCountyVotes(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsCountyVote(maxRequests);
    if (maxed) {
      this.logger.warning('Maximum number of county vote documents returned.');
    }
    if (documents.length === 0) return;

    const updatedVotes = await this.getNewDocuments(
      documents,
      COUNTY_VOTE,
      countyVoteParsers.parseSequenceId
    );

    await this.makeCountyVotes(updatedVotes);

    await this.updateStreamPosition(COUNTY_VOTE, newPosition);

    return {
      /** Whether request was limited by the max number of requests. */
      maxed,
      /** Entities updated via stream */
      updated: updatedVotes.map(({ key }) => {
        const { state, county, jurisdiction, office } = parseKey(key);
        return { state, county, jurisdiction, office };
      }),
    };
  }

  private async makeCalls(documents) {
    const pollCloseTimes = await this.db
      .select({
        jurisdiction: jurisdictionRace.jurisdiction,
        electionType: jurisdictionRace.electionType,
        majorPollCloseTime: jurisdictionRace.majorPollCloseTime,
      })
      .from(jurisdictionRace);

    const calls = estParsers.parseCalls(documents, pollCloseTimes);

    const reversedCalls = calls.filter((call) => call.status === 'R');

    const candidateCalls = estParsers.parseCandidateCalls(
      documents,
      pollCloseTimes
    );
    const documentSequences =
      countyVoteParsers.parseDocumentSequences(documents);
    await this.db.transaction(async (tx) => {
      await this.query.bulkUpsert(call, calls, tx);
      try {
        // Reset call fields to null for reversals
        for (const reversedCall of reversedCalls) {
          // Lookup candidates by CandidateRace_Office Model...
          const { office } = reversedCall;
          const candidateOffices = await tx
            .select({
              candidateRace: candidateRace_office.candidateRace,
            })
            .from(candidateRace_office)
            .where(eq(candidateRace_office.office, office));

          for (const candidateOffice of candidateOffices) {
            const uid = candidateOffice.candidateRace;
            // Update CandidateRace to set null fields.
            await tx
              .update(candidateRace)
              .set({
                winner: null,
                finishingOrder: null,
                candidatePercentage: null,
                advancesToRunoff: null,
                tiedForFirstPlace: null,
                callStatus: null,
                callDescription: null,
              })
              .where(eq(candidateRace.uid, uid));
          }
        }

        // Update CandidateRace with call fields for races that have been actually
        // called, NOT including embargoed calls.
        for (const candidate of candidateCalls) {
          const {
            uid,
            winner,
            finishingOrder,
            candidatePercentage,
            advancesToRunoff,
            tiedForFirstPlace,
            callStatus,
            callDescription,
          } = candidate;
          await tx
            .update(candidateRace)
            .set({
              winner,
              finishingOrder,
              candidatePercentage,
              advancesToRunoff,
              tiedForFirstPlace,
              callStatus,
              callDescription,
            })
            .where(eq(candidateRace.uid, uid));
        }

        await this.query.bulkUpsert(documentSequence, documentSequences, tx);
      } catch {
        throw new DBTransactionError('Error updating calls');
      }
    });
  }

  async updateCalls() {
    const positions = await this.getStreamPositions([EST]);
    const calls = await this.getResultsEst();

    await this.makeCalls(calls);
    await this.query.bulkUpsert(streamPosition, positions);
    await this.updateSupersededStreamPositionsFromSweep(positions);
  }

  async updateStreamedCalls(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsEst(maxRequests);
    if (maxed) {
      this.logger.warning('Maximum number of estimate documents returned.');
    }
    if (documents.length === 0) return;

    const updatedCalls = await this.getNewDocuments(
      documents,
      EST,
      estParsers.parseSequenceId
    );

    await this.makeCalls(updatedCalls);

    await this.updateStreamPosition(EST, newPosition);

    return {
      maxed,
      updated: updatedCalls.map(({ key }) => {
        const { state, jurisdiction, office } = parseKey(key);
        return { state, jurisdiction, office };
      }),
    };
  }

  private async makeElectoralVotes(votes) {
    const electoralVotesByState =
      electoralVoteParsers.parseElectoralVotes(votes);
    const candidateElectoralVotes =
      electoralVoteParsers.parseCandidateElectoralVotes(votes);
    const documentSequences =
      electoralVoteParsers.parseDocumentSequences(votes);

    await this.query.bulkUpsert(electoralVote, electoralVotesByState);
    await this.query.bulkUpsert(
      candidateElectoralVote,
      candidateElectoralVotes
    );
    await this.query.bulkUpsert(documentSequence, documentSequences);
  }

  async updateElectoralVotes() {
    const positions = await this.getStreamPositions([ELECTORAL_VOTE]);
    const votes = await this.getResultsElectoralVote();

    await this.makeElectoralVotes(votes);
    await this.query.bulkUpsert(streamPosition, positions);
    await this.updateSupersededStreamPositionsFromSweep(positions);
  }

  async updateStreamedElectoralVotes(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsElectoralVote(maxRequests);
    if (maxed) {
      this.logger.warning(
        'Maximum number of electoral vote documents returned.'
      );
    }
    if (documents.length === 0) return;

    const updatedVotes = await this.getNewDocuments(
      documents,
      ELECTORAL_VOTE,
      electoralVoteParsers.parseSequenceId
    );

    await this.makeElectoralVotes(updatedVotes);

    await this.updateStreamPosition(ELECTORAL_VOTE, newPosition);

    return {
      maxed,
      updated: updatedVotes.map(({ key }) => {
        const { state, jurisdiction, office } = parseKey(key);
        return { state, jurisdiction, office };
      }),
    };
  }

  private async makeMessages(documents) {
    const messages = msgParsers.parseMessages(documents);
    const documentSequences = msgParsers.parseDocumentSequences(documents);
    await this.db.transaction(async (tx) => {
      await this.query.bulkUpsert(message, messages, tx);
      await this.query.bulkUpsert(documentSequence, documentSequences, tx);
    });
    return messages;
  }

  async updateMessages() {
    const positions = await this.getStreamPositions([MSG]);
    const messages = await this.getResultsMsg();
    await this.makeMessages(messages);
    await this.query.bulkUpsert(streamPosition, positions);
    await this.updateSupersededStreamPositionsFromSweep(positions);
  }

  async updateStreamedMessages(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsMsg(maxRequests);
    if (maxed) {
      this.logger.warning('Maximum number of message documents returned.');
    }
    if (documents.length === 0) return;

    const updatedMessages = await this.getNewDocuments(
      documents,
      MSG,
      msgParsers.parseSequenceId
    );

    const newMessages = await this.makeMessages(updatedMessages);

    await this.updateStreamPosition(MSG, newPosition);

    return {
      maxed,
      updated: newMessages,
    };
  }

  private async makeEditorialTurnouts(documents) {
    const editorialTurnouts =
      editorialTurnoutParsers.parseEditorialTurnouts(documents);
    const documentSequences =
      editorialTurnoutParsers.parseDocumentSequences(documents);

    await this.db.transaction(async (tx) => {
      await this.query.bulkUpsert(editorialTurnout, editorialTurnouts, tx);
      await this.query.bulkUpsert(documentSequence, documentSequences, tx);
    });
    return editorialTurnouts;
  }

  async updateEditorialTurnouts() {
    const positions = await this.getStreamPositions([EDITORIAL_TURNOUT]);
    const editorialTurnouts = await this.getResultsEditorialTurnout();
    await this.makeEditorialTurnouts(editorialTurnouts);
    await this.query.bulkUpsert(streamPosition, positions);
    await this.updateSupersededStreamPositionsFromSweep(positions);
  }

  async updateStreamedEditorialTurnouts(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsEditorialTurnout(maxRequests);
    if (maxed) {
      this.logger.warning(
        'Maximum number of editorial turnout documents returned.'
      );
    }

    if (documents.length === 0) return;

    const updateEditorialTurnouts = await this.getNewDocuments(
      documents,
      EDITORIAL_TURNOUT,
      editorialTurnoutParsers.parseSequenceId
    );

    const newEditorialTurnouts = await this.makeEditorialTurnouts(
      updateEditorialTurnouts
    );

    await this.updateStreamPosition(EDITORIAL_TURNOUT, newPosition);

    return {
      maxed,
      updated: newEditorialTurnouts.map(({ state, jurisdiction }) => ({
        state,
        jurisdiction,
      })),
    };
  }

  private async makeTotalExpectedVotes(documents) {
    const totalExpectedVotes =
      totalExpectedVoteParsers.parseTotalExpectedVotes(documents);
    const documentSequences =
      totalExpectedVoteParsers.parseDocumentSequences(documents);

    await this.db.transaction(async (tx) => {
      await this.query.bulkUpsert(totalExpectedVote, totalExpectedVotes, tx);
      await this.query.bulkUpsert(documentSequence, documentSequences, tx);
    });
    return totalExpectedVotes;
  }

  async updateTotalExpectedVotes() {
    const positions = await this.getStreamPositions([TOTAL_EXPECTED_VOTE]);
    const totalExpectedVotes = await this.getResultsTotalExpectedVote();
    await this.makeTotalExpectedVotes(totalExpectedVotes);
    await this.query.bulkUpsert(streamPosition, positions);
    await this.updateSupersededStreamPositionsFromSweep(positions);
  }

  async updateStreamedTotalExpectedVotes(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsTotalExpectedVote(maxRequests);
    if (maxed) {
      this.logger.warning(
        'Maximum number of total expected vote documents returned.'
      );
    }
    if (documents.length === 0) return;

    /**
     * Total expected vote documents DO NOT contain a key and, according to the schema,
     * may not include a sequence order number, so we add both with the elements
     * that make up the UID we create for the document and a zero sequence order, when
     * undefined.
     *
     * This is only needed in the stream so we filter the most up-to-date docs.
     */
    type TEVDocument = (typeof documents)[number];
    interface TEVDocumentWithKeyAndSeq extends TEVDocument {
      key: GenericDocumentKey;
      seq: number;
    }
    const documentsWithKeyAndSeq: TEVDocumentWithKeyAndSeq[] = [];

    for (const doc of documents) {
      const documentWithKeyAndSeq = {
        ...doc,
        key: {
          eType: doc.eType,
          st: doc.st,
          cnty: doc.cnty,
          jCde: doc.jCde,
          jType: doc.jType,
        },
        /**
         * Force undefined seq to 0, which then *could* be overriden by
         * anything with a later sequence.
         *
         * FWIW, we don't see any docs without seq in tests of the stream, 2020+...
         */
        seq: doc.seq || 0,
      };
      documentsWithKeyAndSeq.push(documentWithKeyAndSeq);
    }

    const updatedTotalExpectedVotes = await this.getNewDocuments(
      documentsWithKeyAndSeq,
      TOTAL_EXPECTED_VOTE,
      totalExpectedVoteParsers.parseSequenceId
    );

    const newTotalExpectedVotes = await this.makeTotalExpectedVotes(
      updatedTotalExpectedVotes
    );

    await this.updateStreamPosition(TOTAL_EXPECTED_VOTE, newPosition);

    return {
      maxed,
      updated: newTotalExpectedVotes,
    };
  }

  private async makeDelegateAllocations(documents) {
    const delegateAllocations =
      delegateAllocationParsers.parseDelegateAllocations(documents);

    await this.query.bulkUpsert(delegateAllocation, delegateAllocations);
    return delegateAllocations;
  }

  async updateDelegateAllocations() {
    const delegateAllocations = await this.getResultsDelegateAllocation();
    await this.makeDelegateAllocations(delegateAllocations);
  }

  async updateStreamedDelegateAllocations(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsDelegateAllocation(maxRequests);
    if (maxed) {
      this.logger.warning(
        'Maximum number of delegate allocation documents returned.'
      );
    }
    if (documents.length === 0) return;

    const updatedDelegateAllocations = await this.getNewDocuments(
      documents,
      DELEGATE_ALLOCATION,
      delegateAllocationParsers.parseSequenceId
    );

    const newDelegateAllocations = await this.makeDelegateAllocations(
      updatedDelegateAllocations
    );

    await this.updateStreamPosition(DELEGATE_ALLOCATION, newPosition);

    return {
      maxed,
      updated: newDelegateAllocations.map(({ state, jurisdiction }) => ({
        state,
        jurisdiction,
      })),
    };
  }

  private async makeDelegateVotes(documents) {
    const presidentialPrimaryVotesByDistrict =
      delegateVoteParsers.parsePresidentialPrimaryVotesByDistrict(documents);

    const presidentialPrimaryDistrictVotesByCounty =
      delegateVoteParsers.parsePresidentialPrimaryDistrictVotesByCounty(
        documents
      );

    await this.db.transaction(async (tx) => {
      await this.query.bulkUpsert(
        presidentialPrimaryVoteByDistrict,
        presidentialPrimaryVotesByDistrict,
        tx
      );
      await this.query.bulkUpsert(
        presidentialPrimaryDistrictVoteByCounty,
        presidentialPrimaryDistrictVotesByCounty,
        tx
      );
    });

    return presidentialPrimaryVotesByDistrict;
  }

  async updateDelegateVotes() {
    const delegateVotes = await this.getResultsDelegateVote();
    await this.makeDelegateVotes(delegateVotes);
  }

  async updateStreamedDelegateVotes(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsDelegateVote(maxRequests);
    if (maxed) {
      this.logger.warning(
        'Maximum number of delegate vote documents returned.'
      );
    }
    if (documents.length === 0) return;

    const updatedDelegateVotes = await this.getNewDocuments(
      documents,
      DELEGATE_VOTE,
      delegateVoteParsers.parseSequenceId
    );

    const newDelegateVotes = await this.makeDelegateVotes(updatedDelegateVotes);

    await this.updateStreamPosition(DELEGATE_VOTE, newPosition);

    return {
      maxed,
      updated: newDelegateVotes,
    };
  }

  private async makeNationalDelegateSummary(documents) {
    const nationalDelegateSummaries =
      delegateSummaryParsers.parseNationalDelegateSummary(documents);
    const stateDelegateSummaries =
      delegateSummaryParsers.parseStateDelegateSummary(documents);

    await this.db.transaction(async (tx) => {
      await this.query.bulkUpsert(
        nationalDelegateSummary,
        nationalDelegateSummaries,
        tx
      );
      await this.query.bulkUpsert(
        stateDelegateSummary,
        stateDelegateSummaries,
        tx
      );
    });

    return nationalDelegateSummary;
  }

  async updateNationalDelegateSummary() {
    const nationalDelegateSummary =
      await this.getResultsNationalDelegateSummary();

    await this.makeNationalDelegateSummary(nationalDelegateSummary);
  }

  async updateStreamedNationalDelegateSummmaries(maxRequests = Infinity) {
    const { documents, newPosition, maxed } =
      await this.getStreamsNationalDelegateSummary(maxRequests);
    if (maxed) {
      this.logger.warning(
        'Maximum number of national delegate summary documents returned.'
      );
    }
    if (documents.length === 0) return;

    const updatedNationalDelegateSummary = await this.getNewDocuments(
      documents,
      NATIONAL_DELEGATE_SUMMARY,
      // @ts-expect-error: TODO fix this...
      delegateSummaryParsers.parseNationalDelegateSummary
    );

    // Updates both NationalDelegateSummary and StateDelegateSummary models
    const newNationalDelegateSummary = await this.makeNationalDelegateSummary(
      updatedNationalDelegateSummary
    );

    await this.updateStreamPosition(NATIONAL_DELEGATE_SUMMARY, newPosition);

    return {
      maxed,
      updated: newNationalDelegateSummary,
    };
  }
}

[{"file":"src/results/index.ts","line":227,"character":49,"text":"positions","kind":1},{"file":"src/results/index.ts","line":232,"character":15,"text":"position","kind":1},{"file":"src/results/index.ts","line":233,"character":14,"text":"docType","kind":1},{"file":"src/results/index.ts","line":233,"character":23,"text":"maxPosition","kind":1},{"file":"src/results/index.ts","line":232,"character":27,"text":"positions","kind":1},{"file":"src/results/index.ts","line":288,"character":32,"text":"votes","kind":1},{"file":"src/results/index.ts","line":341,"character":26,"text":"documents","kind":1},{"file":"src/results/index.ts","line":462,"character":35,"text":"votes","kind":1},{"file":"src/results/index.ts","line":516,"character":29,"text":"documents","kind":1},{"file":"src/results/index.ts","line":558,"character":38,"text":"documents","kind":1},{"file":"src/results/index.ts","line":611,"character":39,"text":"documents","kind":1},{"file":"src/results/index.ts","line":696,"character":40,"text":"documents","kind":1},{"file":"src/results/index.ts","line":740,"character":34,"text":"documents","kind":1},{"file":"src/results/index.ts","line":796,"character":44,"text":"documents","kind":1}]

TypeScript coverage report