POC/Provisioning: Use watch api in the frontend (#98645)

* watch events

* with watch

* Move watch to onCacheEntryAdded

* Update watch

* Fix Badge link

---------

Co-authored-by: Clarity-89 <homes89@ukr.net>

packages/grafana-runtime/src/services/backendSrv.ts

@@ -68,7 +68,7 @@ export type BackendSrvRequest = {
    *
    * By default values are json parsed from text
    */
-  responseType?: 'json' | 'text' | 'arraybuffer' | 'blob';
+  responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream';
 
   /**
    * The credentials read-only property of the Request interface indicates whether the user agent should send cookies from the other domain in the case of cross-origin requests.

public/app/core/utils/fetch.ts

@@ -114,7 +114,7 @@ export const parseBody = (options: BackendSrvRequest, isAppJson: boolean) => {
 
 export async function parseResponseBody<T>(
   response: Response,
-  responseType?: 'json' | 'text' | 'arraybuffer' | 'blob'
+  responseType?: 'json' | 'text' | 'arraybuffer' | 'blob' | 'stream'
 ): Promise<T> {
   if (responseType) {
     switch (responseType) {
@@ -135,9 +135,11 @@ export async function parseResponseBody<T>(
           console.warn(`${response.url} returned an invalid JSON`);
           return {} as T;
         }
-
         return await response.json();
 
+      case 'stream':
+        return Promise.resolve(response.body) as Promise<T>
+
       case 'text':
         // this specifically returns a Promise<string>
         // TODO refactor this function to remove the type assertions

public/app/features/apiserver/client.ts

@@ -1,3 +1,5 @@
+import { Observable, Subscriber, map, switchMap, from, filter, retry, catchError } from 'rxjs';
+
 import { config, getBackendSrv } from '@grafana/runtime';
 import { contextSrv } from 'app/core/core';
 
@@ -13,6 +15,7 @@ import {
   ObjectMeta,
   K8sAPIGroupList,
   AnnoKeySavedFromUI,
+  ResourceEvent,
 } from './types';
 
 export interface GroupVersionResource {
@@ -34,6 +37,107 @@ export class ScopedResourceClient<T = object, S = object, K = string> implements
     return getBackendSrv().get<Resource<T, S, K>>(`${this.url}/${name}`);
   }
 
+  public watch(name?: string, resourceVersion?: string): Observable<ResourceEvent<T, S, K>> {
+    return this.createWatchRequest(name, resourceVersion).pipe(
+      switchMap((response) => this.handleWatchStream(response)),
+      retry({ count: 3, delay: 1000 }),
+      catchError((error) => {
+        console.error('Watch stream error:', error);
+        throw error;
+      })
+    );
+  }
+
+  private createWatchRequest(name?: string, resourceVersion?: string) {
+    return getBackendSrv().fetch<ReadableStream<Uint8Array>>({
+      url: name ? `${this.url}/${name}` : this.url,
+      params: {
+        watch: true,
+        resourceVersion,
+      },
+      responseType: 'stream',
+    });
+  }
+
+  private handleWatchStream(result: { data: ReadableStream<Uint8Array> }): Observable<ResourceEvent<T, S, K>> {
+    const decoder = new TextDecoder();
+    let buffer = '';
+
+    return fromReadableStream(result.data).pipe(
+      map((chunk) => decoder.decode(chunk, { stream: true })),
+      map((text) => {
+        buffer += text;
+        const events: Array<ResourceEvent<T, S, K>> = [];
+
+        // Find complete JSON objects in the buffer
+        let startIndex = buffer.indexOf('{');
+        while (startIndex !== -1) {
+          try {
+            const endIndex = this.findJsonEnd(buffer, startIndex);
+            if (endIndex === -1) {
+              break;
+            }
+
+            const jsonStr = buffer.substring(startIndex, endIndex + 1);
+            const event = JSON.parse(jsonStr) as ResourceEvent<T, S, K>;
+            events.push(event);
+
+            // Remove processed data from buffer
+            buffer = buffer.substring(endIndex + 1);
+            startIndex = buffer.indexOf('{');
+          } catch (e) {
+            // If JSON.parse fails, move to next potential object
+            buffer = buffer.substring(startIndex + 1);
+            startIndex = buffer.indexOf('{');
+          }
+        }
+
+        return events;
+      }),
+      filter((events) => events.length > 0),
+      switchMap((events) => from(events))
+    );
+  }
+
+  private findJsonEnd(str: string, startIndex: number): number {
+    let brackets = 0;
+    let inString = false;
+    let escapeNext = false;
+
+    for (let i = startIndex; i < str.length; i++) {
+      const char = str[i];
+
+      if (escapeNext) {
+        escapeNext = false;
+        continue;
+      }
+
+      if (char === '\\' && inString) {
+        escapeNext = true;
+        continue;
+      }
+
+      if (char === '"' && !escapeNext) {
+        inString = !inString;
+        continue;
+      }
+
+      if (!inString) {
+        if (char === '{') {
+          brackets++;
+        }
+        if (char === '}') {
+          brackets--;
+          if (brackets === 0) {
+            return i;
+          }
+        }
+      }
+    }
+
+    return -1;
+  }
+
   public async subresource<S>(name: string, path: string): Promise<S> {
     return getBackendSrv().get<S>(`${this.url}/${name}/${path}`);
   }
@@ -138,3 +242,48 @@ export const parseListOptionsSelector = (selector: ListOptionsLabelSelector | Li
     .filter(Boolean)
     .join(',');
 };
+
+/**
+ * Creates an Observable source from a ReadableStream source that will emit any
+ * values emitted by the stream.
+ *
+ * https://github.com/rxjs-ninja/rxjs-ninja/blob/main/libs/rxjs/utility/src/lib/from-readable-stream.ts
+ */
+function fromReadableStream<T extends unknown>(
+  stream: ReadableStream<T>,
+  signal?: AbortSignal,
+  queueStrategy?: QueuingStrategy,
+  throwEndAsError = false
+): Observable<T> {
+  function createStream(subscriber: Subscriber<T>) {
+    return new WritableStream<T>(
+      {
+        write: (value) => subscriber.next(value),
+        abort: (error) => {
+          if (throwEndAsError) {
+            subscriber.error(error);
+          } else if (!subscriber.closed) {
+            subscriber.complete();
+          }
+        },
+        close: () => {
+          if (!subscriber.closed) {
+            subscriber.complete();
+          }
+        },
+      },
+      queueStrategy
+    );
+  }
+
+  return new Observable<T>((subscriber) => {
+    stream
+      .pipeTo(createStream(subscriber), { signal })
+      .then(() => {
+        return !subscriber.closed && subscriber.complete();
+      })
+      .catch((error) => subscriber.error(error));
+
+    return () => !stream.locked && stream.cancel();
+  });
+}

public/app/features/apiserver/types.ts

@@ -8,6 +8,8 @@
  *
  */
 
+import { Observable } from 'rxjs';
+
 /** The object type and version */
 export interface TypeMeta<K = string> {
   apiVersion: string;
@@ -169,9 +171,15 @@ export interface MetaStatus {
   details?: object;
 }
 
+export interface ResourceEvent<T = object, S = object, K = string> {
+  type: 'ADDED' | 'DELETED' | 'MODIFIED';
+  object: Resource<T, S, K>;
+}
+
 export interface ResourceClient<T = object, S = object, K = string> {
   create(obj: ResourceForCreate<T, K>): Promise<Resource<T, S, K>>;
   get(name: string): Promise<Resource<T, S, K>>;
+  watch(name?: string, resourceVersion?: string): Observable<ResourceEvent<T, S, K>>;
   subresource<S>(name: string, path: string): Promise<S>;
   list(opts?: ListOptions): Promise<ResourceList<T, S, K>>;
   update(obj: ResourceForCreate<T, K>): Promise<Resource<T, S, K>>;

public/app/features/provisioning/RepositoryListPage.tsx

@@ -18,12 +18,13 @@ import { Page } from 'app/core/components/Page/Page';
 
 import { DeleteRepositoryButton } from './DeleteRepositoryButton';
 import { SyncRepository } from './SyncRepository';
-import { Repository, useGetRepositoryStatusQuery } from './api';
+import { Repository } from './api';
 import { NEW_URL, PROVISIONING_URL } from './constants';
 import { useRepositoryList } from './hooks';
 
 export default function RepositoryListPage() {
   const [items, isLoading] = useRepositoryList();
+
   return (
     <Page navId="provisioning" subTitle="View and manage your configured repositories">
       <Page.Contents isLoading={isLoading}>
@@ -91,7 +92,6 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) {
                     </TextLink>
                   );
                 }
-
                 break;
 
               case 'local':
@@ -106,7 +106,7 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) {
                 </Card.Figure>
                 <Card.Heading>
                   <Stack>
-                    {item.spec?.title} {name && <StatusBadge name={name} />}
+                    {item.spec?.title} <StatusBadge repo={item} name={name} />
                   </Stack>
                 </Card.Heading>
                 <Card.Description>{item.spec?.description}</Card.Description>
@@ -121,8 +121,6 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) {
                   {healthy && <SyncRepository repository={item} />}
                 </Card.Actions>
                 <Card.SecondaryActions>
-                  {/* <IconButton key="comment-alt" name="comment-alt" tooltip="Tooltip content" />
-                  <IconButton key="copy" name="copy" tooltip="Tooltip content" /> */}
                   <DeleteRepositoryButton name={name} />
                 </Card.SecondaryActions>
               </Card>
@@ -136,10 +134,12 @@ function RepositoryListPageContent({ items }: { items?: Repository[] }) {
   );
 }
 
-function StatusBadge({ name }: { name: string }) {
-  const statusQuery = useGetRepositoryStatusQuery({ name }, { pollingInterval: 5000 });
-
-  const state = statusQuery.data?.status?.sync?.state;
+interface StatusBadgeProps {
+  repo: Repository;
+  name: string;
+}
+function StatusBadge({ repo, name }: StatusBadgeProps) {
+  const state = repo.status?.sync?.state;
 
   if (!state) {
     return null;

public/app/features/provisioning/api/endpoints.ts

@@ -1,3 +1,7 @@
+import { Subscription } from 'rxjs';
+
+import { ScopedResourceClient } from '../../apiserver/client';
+
 import { baseAPI as api } from './baseAPI';
 export const addTagTypes = ['Job', 'Repository'] as const;
 const injectedRtkApi = api
@@ -7,7 +11,7 @@ const injectedRtkApi = api
   .injectEndpoints({
     endpoints: (build) => ({
       listJob: build.query<ListJobResponse, ListJobArg | void>({
-        query: (queryArg) => ({
+        query: () => ({
           url: `/jobs`,
         }),
         providesTags: ['Job'],
@@ -22,9 +26,60 @@ const injectedRtkApi = api
         providesTags: ['Job'],
       }),
       listRepository: build.query<ListRepositoryResponse, ListRepositoryArg | void>({
-        query: (queryArg) => ({
+        query: () => ({
           url: `/repositories`,
         }),
+        async onCacheEntryAdded(arg, { updateCachedData, cacheDataLoaded, cacheEntryRemoved }) {
+          const client = new ScopedResourceClient<RepositorySpec, RepositoryStatus>({
+            group: 'provisioning.grafana.app',
+            version: 'v0alpha1',
+            resource: 'repositories',
+          });
+          let subscription: Subscription | null = null;
+          try {
+            // wait for the initial query to resolve before proceeding
+            const response = await cacheDataLoaded;
+            const resourceVersion = response.data.metadata?.resourceVersion;
+
+            subscription = client.watch(undefined, resourceVersion).subscribe((event) => {
+              updateCachedData((draft: RepositoryList) => {
+                if (!draft.items) {
+                  draft.items = [];
+                }
+
+                const existingIndex = draft.items.findIndex(
+                  (item) => item.metadata?.name === event.object.metadata.name
+                );
+
+                if (event.type === 'ADDED') {
+                  // Add the new item
+                  //@ts-expect-error TODO Fix types
+                  draft.items.push(event.object);
+                } else if (event.type === 'MODIFIED') {
+                  // Update the existing item if it exists
+                  if (existingIndex !== -1) {
+                    //@ts-expect-error TODO Fix types
+                    draft.items[existingIndex] = event.object;
+                  }
+                } else if (event.type === 'DELETED') {
+                  // Remove the item if it exists
+                  if (existingIndex !== -1) {
+                    draft.items.splice(existingIndex, 1);
+                  }
+                }
+              });
+            });
+          } catch (error) {
+            // no-op in case `cacheEntryRemoved` resolves before `cacheDataLoaded`,
+            // in which case `cacheDataLoaded` will throw
+            console.error('Error in onCacheEntryAdded:', error);
+          }
+          // cacheEntryRemoved will resolve when the cache subscription is no longer active
+          await cacheEntryRemoved;
+          // perform cleanup steps once the `cacheEntryRemoved` promise resolves
+
+          subscription?.unsubscribe();
+        },
         providesTags: ['Repository'],
       }),
       createRepository: build.mutation<CreateRepositoryResponse, CreateRepositoryArg>({