Merge branch 'main' into better-flowable-flow-log

ui/app/api/mirrors/alerts/route.ts

@@ -0,0 +1,21 @@
+import prisma from '@/app/utils/prisma';
+
+export const dynamic = 'force-dynamic';
+
+export async function POST(request: Request) {
+  const { flowName } = await request.json();
+  const errCount = await prisma.flow_errors.count({
+    where: {
+      flow_name: flowName,
+      error_type: 'error',
+      ack: false,
+    },
+  });
+  let mirrorStatus: 'healthy' | 'failed';
+  if (errCount > 0) {
+    mirrorStatus = 'failed';
+  } else {
+    mirrorStatus = 'healthy';
+  }
+  return new Response(JSON.stringify(mirrorStatus));
+}

ui/app/dto/MirrorsDTO.ts

@@ -28,3 +28,12 @@ export type SyncStatusRow = {
   endTime: Date | null;
   numRows: number;
 };
+
+export type AlertErr = {
+  id: bigint;
+  flow_name: string;
+  error_message: string;
+  error_type: string;
+  error_timestamp: Date;
+  ack: boolean;
+};

ui/app/mirrors/errors/[mirrorName]/page.tsx

@@ -0,0 +1,75 @@
+import { AlertErr } from '@/app/dto/MirrorsDTO';
+import prisma from '@/app/utils/prisma';
+import TimeLabel from '@/components/TimeComponent';
+import { Label } from '@/lib/Label';
+import { Table, TableCell, TableRow } from '@/lib/Table';
+
+type MirrorErrorProps = {
+  params: { mirrorName: string };
+};
+
+const MirrorError = async ({ params: { mirrorName } }: MirrorErrorProps) => {
+  const mirrorErrors: AlertErr[] = await prisma.flow_errors.findMany({
+    where: {
+      flow_name: mirrorName,
+      error_type: 'error',
+    },
+    distinct: ['error_message'],
+  });
+
+  return (
+    <div style={{ padding: '2rem' }}>
+      <Label variant='title2'>Error Log</Label>
+      <hr></hr>
+      <div style={{ marginTop: '1rem' }}>
+        <Label variant='body'>
+          <b>Mirror name</b>:
+        </Label>
+        <Label variant='body'>{mirrorName}</Label>
+        <div
+          style={{
+            fontSize: 15,
+            marginTop: '1rem',
+            width: '100%',
+            border: '1px solid rgba(0,0,0,0.1)',
+            padding: '1rem',
+            borderRadius: '1rem',
+          }}
+        >
+          <Table
+            header={
+              <TableRow style={{ textAlign: 'left' }}>
+                <TableCell>Type</TableCell>
+                <TableCell>Message</TableCell>
+                <TableCell>
+                  <Label as='label' style={{ fontSize: 15 }}>
+                    Timestamp
+                  </Label>
+                </TableCell>
+              </TableRow>
+            }
+          >
+            {mirrorErrors.map((mirrorError) => (
+              <TableRow key={mirrorError.error_message}>
+                <TableCell style={{ color: '#F45156', width: '10%' }}>
+                  {mirrorError.error_type.toUpperCase()}
+                </TableCell>
+                <TableCell style={{ width: '70%', fontSize: 13 }}>
+                  {mirrorError.error_message}
+                </TableCell>
+                <TableCell style={{ width: '30%' }}>
+                  <TimeLabel
+                    fontSize={14}
+                    timeVal={mirrorError.error_timestamp.toLocaleString()}
+                  />
+                </TableCell>
+              </TableRow>
+            ))}
+          </Table>
+        </div>
+      </div>
+    </div>
+  );
+};
+
+export default MirrorError;

ui/app/mirrors/mirror-status.tsx

@@ -0,0 +1,83 @@
+'use client';
+
+import { Button } from '@/lib/Button';
+import { Icon } from '@/lib/Icon';
+import { Label } from '@/lib/Label';
+import { ProgressCircle } from '@/lib/ProgressCircle';
+import Link from 'next/link';
+import { useRouter } from 'next/navigation';
+import { useEffect, useState } from 'react';
+export const ErrorModal = ({ flowName }: { flowName: string }) => {
+  const router = useRouter();
+  return (
+    <Link href={`/mirrors/errors/${flowName}`}>
+      <Button
+        style={{
+          backgroundColor: 'rgba(240, 128, 128, 0.5)',
+          height: '2rem',
+          boxShadow: '0 2px 4px rgba(0, 0, 0, 0.2)',
+        }}
+      >
+        <Label as='label' style={{ fontSize: 13, color: 'darkred' }}>
+          Show errors
+        </Label>
+      </Button>
+    </Link>
+  );
+};
+
+export const MirrorError = ({ flowName }: { flowName: string }) => {
+  const [flowStatus, setFlowStatus] = useState<string>();
+  const [isLoading, setIsLoading] = useState<boolean>(true);
+  const [error, setError] = useState<string | null>(null);
+
+  useEffect(() => {
+    const fetchData = async () => {
+      setIsLoading(true);
+      try {
+        const response = await fetch(`/api/mirrors/alerts`, {
+          method: 'POST',
+          headers: {
+            'Content-Type': 'application/json',
+          },
+          body: JSON.stringify({ flowName }),
+        });
+
+        if (!response.ok) {
+          throw new Error('Network response was not ok');
+        }
+
+        const flowStatus = await response.json();
+        setFlowStatus(flowStatus);
+      } catch (err: any) {
+        setError(err.message);
+      } finally {
+        setIsLoading(false);
+      }
+    };
+
+    fetchData();
+  }, [flowName]);
+
+  if (isLoading) {
+    return (
+      <div>
+        <ProgressCircle variant='intermediate_progress_circle' />
+      </div>
+    );
+  }
+
+  if (error) {
+    return (
+      <div>
+        <Icon name='error' />
+      </div>
+    );
+  }
+
+  if (flowStatus == 'healthy') {
+    return <Icon name='check_circle' fill={true} />;
+  }
+
+  return <ErrorModal flowName={flowName} />;
+};

ui/app/mirrors/tables.tsx

@@ -7,6 +7,7 @@ import { SearchField } from '@/lib/SearchField';
 import { Table, TableCell, TableRow } from '@/lib/Table';
 import Link from 'next/link';
 import { useMemo, useState } from 'react';
+import { MirrorError } from './mirror-status';
 
 export function CDCFlows({ cdcFlows }: { cdcFlows: any }) {
   const [searchQuery, setSearchQuery] = useState<string>('');
@@ -43,15 +44,26 @@ export function CDCFlows({ cdcFlows }: { cdcFlows: any }) {
           }}
           header={
             <TableRow>
-              {['Name', 'Source', 'Destination', 'Start Time', ''].map(
-                (heading, index) => (
-                  <TableCell as='th' key={index}>
-                    <Label as='label' style={{ fontWeight: 'bold' }}>
-                      {heading}
-                    </Label>
-                  </TableCell>
-                )
-              )}
+              {[
+                'Name',
+                'Source',
+                'Destination',
+                'Start Time',
+                'Status',
+                '',
+              ].map((heading, index) => (
+                <TableCell as='th' key={index}>
+                  <Label
+                    as='label'
+                    style={{
+                      fontWeight: 'bold',
+                      padding: heading === 'Status' ? 0 : 'auto',
+                    }}
+                  >
+                    {heading}
+                  </Label>
+                </TableCell>
+              ))}
             </TableRow>
           }
         >
@@ -77,6 +89,9 @@ export function CDCFlows({ cdcFlows }: { cdcFlows: any }) {
               <TableCell>
                 <TimeLabel timeVal={flow.created_at} />
               </TableCell>
+              <TableCell>
+                <MirrorError flowName={flow.name} />
+              </TableCell>
               <TableCell>
                 <DropDialog
                   mode='MIRROR'

ui/prisma/schema.prisma

@@ -155,7 +155,51 @@ model peer_slot_size {
   confirmed_flush_lsn String?
   slot_size           BigInt?
   updated_at          DateTime @default(now()) @db.Timestamp(6)
+  wal_status          String?
 
   @@index([slot_name], map: "index_slot_name")
   @@schema("peerdb_stats")
 }
+
+/// This table contains check constraints and requires additional setup for migrations. Visit https://pris.ly/d/check-constraints for more info.
+model alerting_config {
+  id             BigInt @id @default(autoincrement())
+  service_type   String
+  service_config Json
+
+  @@schema("peerdb_stats")
+}
+
+/// This table contains check constraints and requires additional setup for migrations. Visit https://pris.ly/d/check-constraints for more info.
+model alerts_v1 {
+  id                BigInt    @id @default(autoincrement())
+  alert_key         String
+  alert_level       String    @default("critical")
+  alert_message     String
+  created_timestamp DateTime? @default(now()) @db.Timestamp(6)
+
+  @@schema("peerdb_stats")
+}
+
+model flow_errors {
+  id              BigInt   @id @default(autoincrement())
+  flow_name       String
+  error_message   String
+  error_type      String
+  error_timestamp DateTime @default(now()) @db.Timestamp(6)
+  ack             Boolean  @default(false)
+
+  @@index([flow_name], map: "idx_flow_errors_flow_name")
+  @@schema("peerdb_stats")
+}
+
+model schema_deltas_audit_log {
+  id             BigInt    @id @default(autoincrement())
+  flow_job_name  String
+  read_timestamp DateTime? @default(now()) @db.Timestamp(6)
+  workflow_id    String
+  run_id         String
+  delta_info     Json
+
+  @@schema("peerdb_stats")
+}