remove device stream implementation

src/schema/api.graphql

@@ -229,6 +229,10 @@ type DeleteSnapshotPayload {
   snapshot: Snapshot
 }
 
+type DeleteStreamPayload {
+  stream: Stream
+}
+
 type Device implements Node {
   address: String
   blueprint: Blueprint
@@ -434,6 +438,7 @@ type Mutation {
   deleteDevice(id: String!): DeleteDevicePayload!
   deleteLabel(id: String!): DeleteLabelPayload!
   deleteSnapshot(input: DeleteSnapshotInput!): DeleteSnapshotPayload
+  deleteStream(id: String!): DeleteStreamPayload!
   importCSV(input: CSVImportInput!): CSVImport
   installDevice(id: String!): InstallDevicePayload!
   reconnectKafka: IsOkResponse

src/schema/nexus-typegen.ts

@@ -325,6 +325,10 @@ export interface NexusGenObjects {
     // root type
     snapshot?: NexusGenRootTypes['Snapshot'] | null; // Snapshot
   };
+  DeleteStreamPayload: {
+    // root type
+    stream?: NexusGenRootTypes['Stream'] | null; // Stream
+  };
   Device: SourceTypes.Device;
   DeviceConnection: {
     // root type
@@ -834,6 +838,10 @@ export interface NexusGenFieldTypes {
     // field return type
     snapshot: NexusGenRootTypes['Snapshot'] | null; // Snapshot
   };
+  DeleteStreamPayload: {
+    // field return type
+    stream: NexusGenRootTypes['Stream'] | null; // Stream
+  };
   Device: {
     // field return type
     address: string | null; // String
@@ -985,6 +993,7 @@ export interface NexusGenFieldTypes {
     deleteDevice: NexusGenRootTypes['DeleteDevicePayload']; // DeleteDevicePayload!
     deleteLabel: NexusGenRootTypes['DeleteLabelPayload']; // DeleteLabelPayload!
     deleteSnapshot: NexusGenRootTypes['DeleteSnapshotPayload'] | null; // DeleteSnapshotPayload
+    deleteStream: NexusGenRootTypes['DeleteStreamPayload']; // DeleteStreamPayload!
     importCSV: NexusGenRootTypes['CSVImport'] | null; // CSVImport
     installDevice: NexusGenRootTypes['InstallDevicePayload']; // InstallDevicePayload!
     reconnectKafka: NexusGenRootTypes['IsOkResponse'] | null; // IsOkResponse
@@ -1448,6 +1457,10 @@ export interface NexusGenFieldTypeNames {
     // field return type name
     snapshot: 'Snapshot';
   };
+  DeleteStreamPayload: {
+    // field return type name
+    stream: 'Stream';
+  };
   Device: {
     // field return type name
     address: 'String';
@@ -1599,6 +1612,7 @@ export interface NexusGenFieldTypeNames {
     deleteDevice: 'DeleteDevicePayload';
     deleteLabel: 'DeleteLabelPayload';
     deleteSnapshot: 'DeleteSnapshotPayload';
+    deleteStream: 'DeleteStreamPayload';
     importCSV: 'CSVImport';
     installDevice: 'InstallDevicePayload';
     reconnectKafka: 'IsOkResponse';
@@ -2008,6 +2022,10 @@ export interface NexusGenArgTypes {
       // args
       input: NexusGenInputs['DeleteSnapshotInput']; // DeleteSnapshotInput!
     };
+    deleteStream: {
+      // args
+      id: string; // String!
+    };
     importCSV: {
       // args
       input: NexusGenInputs['CSVImportInput']; // CSVImportInput!

src/schema/stream.ts

@@ -11,6 +11,7 @@ import {
   uninstallDeviceCache,
 } from '../external-api/uniconfig-cache';
 import { getMountParamsForStream, getUniconfigStreamName } from '../helpers/stream-helpers';
+import config from '../config';
 
 export const StreamNode = objectType({
   name: 'Stream',
@@ -241,3 +242,55 @@ export const DeactivateStreamMutation = extendType({
     });
   },
 });
+
+export const DeleteStreamPayload = objectType({
+  name: 'DeleteStreamPayload',
+  definition: (t) => {
+    t.field('stream', { type: StreamNode });
+  },
+});
+
+export const DeleteStreamMutation = extendType({
+  type: 'Mutation',
+  definition: (t) => {
+    t.nonNull.field('deleteStream', {
+      type: DeleteStreamPayload,
+      args: {
+        id: nonNull(stringArg()),
+      },
+      resolve: async (_, args, { prisma, tenantId, kafka, inventoryKafka }) => {
+        const nativeId = fromGraphId('Stream', args.id);
+        const dbStream = await prisma.stream.findFirst({
+          where: { id: nativeId, AND: { tenantId } },
+          include: { device: true },
+        });
+        if (dbStream == null) {
+          throw new Error('device not found');
+        }
+        const uniconfigURL = await getUniconfigURL(prisma, dbStream.device.uniconfigZoneId);
+        const isActive = await getCachedDeviceInstallStatus(
+          uniconfigURL,
+          getUniconfigStreamName(dbStream.streamName, dbStream.deviceName),
+        );
+        if (isActive) {
+          throw new Error('stream is installed in UniConfig');
+        }
+
+        try {
+          const deletedStream = await prisma.stream.delete({ where: { id: nativeId } });
+
+          if (config.kafkaEnabled) {
+            await inventoryKafka?.produceDeviceRemovalEvent(
+              kafka,
+              getUniconfigStreamName(dbStream.streamName, dbStream.deviceName),
+            );
+          }
+
+          return { stream: deletedStream };
+        } catch (error) {
+          throw new Error('Error deleting stream');
+        }
+      },
+    });
+  },
+});