async function allocateChunk(channelId) {
  const today = moment().utc().startOf('day');
  let newChunkExpiry = null;
  dbg('Allocating channel chunk for today', today.toISOString());

  const channel = await Channel.findById(channelId).select('activeChunk chunkExpiry').lean().exec();
  let activeChunk = await ChannelChunk.findById(channel.activeChunk).select('id').lean().exec();

  if (!activeChunk) {
    dbg(`No chunk for channel #${channelId}. Making one afresh...`);
  } else if (moment(channel.chunkExpiry).isSameOrBefore(today, 'day')) {
    dbg(`Existing for channel #${channelId} has expired. Making one afresh...`);
  }

  if (!activeChunk || (moment(channel.chunkExpiry).isSameOrBefore(today, 'day'))) {
    // NOTICE: Addressing the concurrent update issue
    const result = await ChannelChunk.updateOne(
      {
        channelId,
        createdAt: today,
      }, {}, { upsert: true },
    ).exec();

    if (result.ok && result.upserted && result.upserted.length) {
      activeChunk = { _id: result.upserted[0]._id };
    } else if (result.ok) {
      dbg('Another process had created channel chunk');
      activeChunk = await ChannelChunk.findOne(
        {
          channelId,
          createdAt: today,
        },
      ).select('id').lean().exec();
    }

    newChunkExpiry = today.add(CHUNK_EXPIRE_IN, 'day');
  }

  if (newChunkExpiry) {
    dbg(`Updating active chunk #${activeChunk._id} and chunkExpiry for channel #${channelId}...`);
    await Channel.findByIdAndUpdate(channelId,
      {
        $set:
          {
            activeChunk: activeChunk._id,
            chunkExpiry: newChunkExpiry,
          },
      });
  } else {
    dbg(`Reusing existing chunk #${activeChunk._id} for channel #${channelId}...`);
  }

  return activeChunk._id;
}
/**
 * Allocate a new channel chunk for storing new messages
 *
 * @param channelId {string}
 * @returns {Promise<string>}
 */
Allocate a new channel chunk for storing new messages
@param channelId {string}
@returns {Promise}
Allocate a new channel chunk for storing new messages @param channelId { string } @returns { Promise }
async function allocateChunk(channelId) {
  const today = moment().utc().startOf('day');
  let newChunkExpiry = null;
  dbg('Allocating channel chunk for today', today.toISOString());
  const channel = await Channel.findById(channelId).select('activeChunk chunkExpiry').lean().exec();
  let activeChunk = await ChannelChunk.findById(channel.activeChunk).select('id').lean().exec();
  if (!activeChunk) {
    dbg(`No chunk for channel #${channelId}. Making one afresh...`);
  } else if (moment(channel.chunkExpiry).isSameOrBefore(today, 'day')) {
    dbg(`Existing for channel #${channelId} has expired. Making one afresh...`);
  }
  if (!activeChunk || (moment(channel.chunkExpiry).isSameOrBefore(today, 'day'))) {
    const result = await ChannelChunk.updateOne(
      {
        channelId,
        createdAt: today,
      }, {}, { upsert: true },
    ).exec();
    if (result.ok && result.upserted && result.upserted.length) {
      activeChunk = { _id: result.upserted[0]._id };
    } else if (result.ok) {
      dbg('Another process had created channel chunk');
      activeChunk = await ChannelChunk.findOne(
        {
          channelId,
          createdAt: today,
        },
      ).select('id').lean().exec();
    }
    newChunkExpiry = today.add(CHUNK_EXPIRE_IN, 'day');
  }
  if (newChunkExpiry) {
    dbg(`Updating active chunk #${activeChunk._id} and chunkExpiry for channel #${channelId}...`);
    await Channel.findByIdAndUpdate(channelId,
      {
        $set:
          {
            activeChunk: activeChunk._id,
            chunkExpiry: newChunkExpiry,
          },
      });
  } else {
    dbg(`Reusing existing chunk #${activeChunk._id} for channel #${channelId}...`);
  }
  return activeChunk._id;
}
async function allocateChunk ( channelId ) { const today = moment ( ) . utc ( ) . startOf ( 'day' ) ; let newChunkExpiry = null ; dbg ( 'Allocating channel chunk for today' , today . toISOString ( ) ) ; const channel = await Channel . findById ( channelId ) . select ( 'activeChunk chunkExpiry' ) . lean ( ) . exec ( ) ; let activeChunk = await ChannelChunk . findById ( channel . activeChunk ) . select ( 'id' ) . lean ( ) . exec ( ) ; if ( ! activeChunk ) { dbg ( ` ${ channelId } ` ) ; } else if ( moment ( channel . chunkExpiry ) . isSameOrBefore ( today , 'day' ) ) { dbg ( ` ${ channelId } ` ) ; } if ( ! activeChunk || ( moment ( channel . chunkExpiry ) . isSameOrBefore ( today , 'day' ) ) ) { const result = await ChannelChunk . updateOne ( { channelId , createdAt : today , } , { } , { upsert : true } , ) . exec ( ) ; if ( result . ok && result . upserted && result . upserted . length ) { activeChunk = { _id : result . upserted [ 0 ] . _id } ; } else if ( result . ok ) { dbg ( 'Another process had created channel chunk' ) ; activeChunk = await ChannelChunk . findOne ( { channelId , createdAt : today , } , ) . select ( 'id' ) . lean ( ) . exec ( ) ; } newChunkExpiry = today . add ( CHUNK_EXPIRE_IN , 'day' ) ; } if ( newChunkExpiry ) { dbg ( ` ${ activeChunk . _id } ${ channelId } ` ) ; await Channel . findByIdAndUpdate ( channelId , { $set : { activeChunk : activeChunk . _id , chunkExpiry : newChunkExpiry , } , } ) ; } else { dbg ( ` ${ activeChunk . _id } ${ channelId } ` ) ; } return activeChunk . _id ; }
Allocate a new channel chunk for storing new messages
 @param channelId {string}
 @returns {Promise<string>}
Allocate a new channel chunk for storing new messages @param channelId { string } @returns { Promise<string > }
// NOTICE: Addressing the concurrent update issue