Added tiling for pending posts and uploads

src/common/cron.ts

@@ -0,0 +1,3 @@
+export enum FunkyCronExpression {
+  EVERY_3_MINUTES = '*/3 * * * *',
+}

src/common/date/date-buckets.ts

@@ -264,3 +264,31 @@ export const generateSeriesRecordPoints = <R extends Record<string, number>>(
     ),
   }));
 };
+
+export const generateSeriesTileCountPoints = (
+  dates: DatePoint[],
+  counts: number[],
+  range: PartialDateRange,
+): SeriesCountPoint[] => {
+  return generateSeriesPoints(counts, dates, range).map(
+    (e) =>
+      new SeriesCountPoint({
+        date: e.date,
+        value: e.value.reduce((sum, count) => sum + count, 0),
+      }),
+  );
+};
+
+export const generateSeriesLastTileCountPoints = (
+  dates: DatePoint[],
+  counts: number[],
+  range: PartialDateRange,
+): SeriesCountPoint[] => {
+  return generateSeriesPoints(counts, dates, range).map(
+    (e) =>
+      new SeriesCountPoint({
+        date: e.date,
+        value: e.value.length > 0 ? e.value[e.value.length - 1]! : 0,
+      }),
+  );
+};

src/common/index.ts

@@ -1,6 +1,7 @@
+export * from './date';
 export * from './case';
 export * from './chart.dto';
-export * from './date';
+export * from './cron';
 export * from './id-range.dto';
 export * from './limit';
 export * from './logs';
@@ -10,4 +11,5 @@ export * from './raw';
 export * from './repository';
 export * from './request-context';
 export * from './seed';
+export * from './tile';
 export * from './timezone';

src/common/tile.ts

@@ -0,0 +1,209 @@
+import { add, isEqual, min } from 'date-fns';
+import { Repository } from 'typeorm';
+
+import { DateRange, PartialDateRange, TimeScale, startOf } from './date';
+
+export enum TileType {
+  uploadHourly = 'upload_hourly',
+  postPendingHourly = 'post_pending_hourly',
+}
+
+export interface TileService {
+  /**
+   * Interval in hours between tiles of this type.
+   */
+  interval: number;
+
+  /**
+   * Get the available tiling ranges for this tile type.
+   */
+  getRanges: () => Promise<TilingRange[]>;
+
+  /**
+   * Find missing tiles in the given date range.
+   */
+  findMissing: (range: TilingRange) => Promise<Date[]>;
+
+  /**
+   * Wipe all tiles of this type, or tiles within a date range if specified.
+   */
+  wipe: (range?: PartialDateRange) => Promise<void>;
+}
+
+export interface TilingRange {
+  /**
+   * The date range for the tiles.
+   */
+  dateRange: DateRange;
+
+  /**
+   * The minimum updatedAt timestamp for the tiles in this range.
+   */
+  updatedAt?: Date;
+
+  /**
+   * Type identifier for this range. Undefined when not applicable.
+   */
+  type?: string;
+}
+
+export const getTilingRanges = (
+  manifests: Array<TilingRange>,
+  types: string[],
+): TilingRange[] => {
+  const events: Array<{
+    time: Date;
+    type: string;
+    delta: number;
+    updatedAt?: Date;
+  }> = [];
+
+  for (const manifest of manifests) {
+    const type = manifest.type ?? 'default';
+    const start = startOf(TimeScale.Hour, manifest.dateRange.startDate);
+    const endFloor = startOf(TimeScale.Hour, manifest.dateRange.endDate);
+    const end =
+      endFloor.getTime() === manifest.dateRange.endDate.getTime()
+        ? endFloor
+        : add(endFloor, { hours: 1 });
+
+    events.push({ time: start, type, delta: 1, updatedAt: manifest.updatedAt });
+    events.push({ time: end, type, delta: -1, updatedAt: manifest.updatedAt });
+  }
+
+  events.sort((a, b) => {
+    const diff = a.time.getTime() - b.time.getTime();
+    if (diff !== 0) return diff;
+    // We want splits in ranges, to make them manageable.
+    // To return one continuous range, this could be flipped.
+    return a.delta - b.delta;
+  });
+
+  const depths = new Map<string, number>();
+
+  const ranges: TilingRange[] = [];
+  let start: Date | null = null;
+  let updated: Date | undefined = undefined;
+
+  for (const event of events) {
+    const prevDepth = depths.get(event.type) ?? 0;
+    const newDepth = prevDepth + event.delta;
+    depths.set(event.type, newDepth);
+
+    const active = types.every((type) => (depths.get(type) ?? 0) > 0);
+
+    if (active) {
+      if (start === null) {
+        start = event.time;
+      }
+      if (event.updatedAt && (!updated || event.updatedAt > updated)) {
+        updated = event.updatedAt;
+      }
+    } else if (start !== null) {
+      ranges.push({
+        dateRange: new DateRange({
+          startDate: start,
+          endDate: event.time,
+          scale: TimeScale.Hour,
+        }),
+        updatedAt: updated,
+      });
+      start = null;
+      updated = undefined;
+    }
+  }
+
+  return ranges;
+};
+
+export const groupTimesIntoRanges = (times: Date[]): DateRange[] => {
+  if (times.length === 0) return [];
+
+  const sorted = [...times].sort((a, b) => a.getTime() - b.getTime());
+  const ranges: DateRange[] = [];
+
+  let start = sorted[0]!;
+  let end = add(sorted[0]!, { hours: 1 });
+
+  for (const current of sorted.slice(1)) {
+    if (isEqual(current, end)) {
+      end = add(current, { hours: 1 });
+    } else {
+      ranges.push(
+        new DateRange({
+          startDate: start,
+          endDate: end,
+          scale: TimeScale.Hour,
+        }),
+      );
+      start = current;
+      end = add(current, { hours: 1 });
+    }
+  }
+
+  ranges.push(
+    new DateRange({
+      startDate: start,
+      endDate: end,
+      scale: TimeScale.Hour,
+    }),
+  );
+
+  return ranges;
+};
+
+export const chunkDateRange = (
+  range: DateRange,
+  maxHours: number,
+): DateRange[] => {
+  const chunks: DateRange[] = [];
+  let current = range.startDate;
+
+  while (current < range.endDate) {
+    const chunkEnd = add(current, { hours: maxHours });
+    const actualEnd = min([chunkEnd, range.endDate]);
+
+    chunks.push(
+      new DateRange({
+        startDate: current,
+        endDate: actualEnd,
+        scale: TimeScale.Hour,
+      }),
+    );
+
+    current = actualEnd;
+  }
+
+  return chunks;
+};
+
+export interface WithTileTime {
+  time: Date;
+  updatedAt: Date;
+}
+
+export async function findMissingOrStaleTiles<T extends WithTileTime>(
+  repository: Repository<T>,
+  range: TilingRange,
+): Promise<Date[]> {
+  const startTime = startOf(TimeScale.Hour, range.dateRange.startDate);
+  const endTime = startOf(TimeScale.Hour, range.dateRange.endDate);
+  const tableName = repository.metadata.tableName;
+
+  const query = `
+    SELECT series.hour AS time
+    FROM generate_series(
+      $1::timestamptz,
+      $2::timestamptz - interval '1 hour',
+      interval '1 hour'
+    ) AS series(hour)
+    LEFT JOIN ${tableName} tile ON tile.time = series.hour
+    WHERE tile.time IS NULL
+       OR ($3::timestamptz IS NOT NULL AND tile.updated_at < $3::timestamptz)
+  `;
+
+  const params = [startTime, endTime, range.updatedAt || null];
+  const result = await repository.query(query, params);
+
+  return result.map((row: { time: Date }) => row.time);
+}

src/health/health.module.ts

@@ -2,9 +2,10 @@ import { Module } from '@nestjs/common';
 
 import { HealthController } from './health.controller';
 import { ManifestHealthModule } from './manifests/manifest-health.module';
+import { TileHealthModule } from './tiles/tile-health.module';
 
 @Module({
-  imports: [ManifestHealthModule],
+  imports: [ManifestHealthModule, TileHealthModule],
   controllers: [HealthController],
 })
 export class HealthModule {}

src/health/tiles/index.ts

@@ -0,0 +1,5 @@
+export * from './tile-health.controller';
+export * from './tile-health.dto';
+export * from './tile-health.module';
+export * from './tile-health.service';
+export * from './tile-health.utils';

src/health/tiles/tile-health.controller.ts

@@ -0,0 +1,71 @@
+import {
+  Controller,
+  Delete,
+  Get,
+  Param,
+  Query,
+  UseGuards,
+} from '@nestjs/common';
+import {
+  ApiBearerAuth,
+  ApiOperation,
+  ApiParam,
+  ApiResponse,
+  ApiTags,
+} from '@nestjs/swagger';
+import { ServerAdminGuard } from 'src/auth/auth.guard';
+import { PartialDateRange, TileType } from 'src/common';
+import { PaginationParams } from 'src/common/pagination.dto';
+
+import { TileHealth } from './tile-health.dto';
+import { TileHealthService } from './tile-health.service';
+
+@ApiTags('Health')
+@Controller('health/tiles')
+export class TileHealthController {
+  constructor(private readonly tileHealthService: TileHealthService) {}
+
+  @Get()
+  @ApiOperation({
+    summary: 'Retrieve tile health',
+    description: 'Retrieve tile health and coverage information',
+    operationId: 'getTileHealth',
+  })
+  @ApiResponse({
+    status: 200,
+    description: 'Tile health information',
+    type: [TileHealth],
+  })
+  @UseGuards(ServerAdminGuard)
+  @ApiBearerAuth()
+  async getTileHealth(
+    @Query() pages?: PaginationParams,
+  ): Promise<TileHealth[]> {
+    return this.tileHealthService.tiles(pages);
+  }
+
+  // This is kind of awkward, being handled in the health controller.
+  @Delete(':type')
+  @ApiOperation({
+    summary: 'Delete all tiles of a type',
+    description: 'Delete all tiles of the specified type',
+    operationId: 'deleteTilesByType',
+  })
+  @ApiParam({
+    name: 'type',
+    enum: TileType,
+    description: 'The type of tiles to delete',
+  })
+  @ApiResponse({
+    status: 204,
+    description: 'Tiles deleted successfully',
+  })
+  @UseGuards(ServerAdminGuard)
+  @ApiBearerAuth()
+  async deleteTilesByType(
+    @Param('type') type: TileType,
+    @Query() range?: PartialDateRange,
+  ): Promise<void> {
+    return this.tileHealthService.deleteTilesByType(type, range);
+  }
+}

src/health/tiles/tile-health.dto.ts

@@ -0,0 +1,29 @@
+import { ApiProperty } from '@nestjs/swagger';
+import { Raw, TileType } from 'src/common';
+
+export class TileSlice {
+  constructor(value: Raw<TileSlice>) {
+    Object.assign(this, value);
+  }
+
+  startDate: Date;
+  endDate: Date;
+
+  available: number;
+  unavailable: number;
+  none: number;
+}
+
+export class TileHealth {
+  constructor(value: Raw<TileHealth>) {
+    Object.assign(this, value);
+  }
+
+  @ApiProperty({ enum: TileType, enumName: 'TileType' })
+  type: TileType;
+  startDate: Date;
+  endDate: Date;
+  expected: number;
+  actual: number;
+  slices: TileSlice[];
+}