konnosDashboard
useRealtimeMessages.ts241 lines · main
1import { RealtimeChannel, RealtimeClient } from '@supabase/realtime-js'
2import { sortBy, take } from 'lodash'
3import { Dispatch, SetStateAction, useCallback, useEffect, useReducer, useState } from 'react'
4import { toast } from 'sonner'
5
6import type { LogData } from './Messages.types'
7import { useProjectSettingsV2Query } from '@/data/config/project-settings-v2-query'
8import { uuidv4 } from '@/lib/helpers'
9import { EMPTY_ARR } from '@/lib/void'
10import { useRoleImpersonationStateSnapshot } from '@/state/role-impersonation-state'
11
12const DEFAULT_HEADERS = { 'X-Client-Info': 'briven-js-web/studio' }
13
14function reducer(
15 state: LogData[],
16 action: { type: 'add'; payload: { messageType: string; metadata: any } } | { type: 'clear' }
17) {
18 if (action.type === 'clear') {
19 return EMPTY_ARR
20 }
21
22 const newState = take(
23 sortBy(
24 [
25 {
26 id: uuidv4(),
27 timestamp: new Date().getTime(),
28 message: action.payload.messageType,
29 metadata: action.payload.metadata,
30 } as LogData,
31 ...state,
32 ],
33 (l) => -l.timestamp
34 ),
35 100
36 )
37
38 return newState
39}
40
41export interface RealtimeConfig {
42 enabled: boolean
43 channelName: string
44 projectRef: string
45 logLevel: string
46 token: string
47 schema: string
48 table: string
49 isChannelPrivate: boolean
50 filter: string | undefined
51 bearer: string | null
52 enableBroadcast: boolean
53 enablePresence: boolean
54 enableDbChanges: boolean
55}
56
57export const useRealtimeMessages = (
58 config: RealtimeConfig,
59 setRealtimeConfig: Dispatch<SetStateAction<RealtimeConfig>>
60) => {
61 const {
62 enabled,
63 channelName,
64 projectRef,
65 logLevel,
66 token,
67 schema,
68 table,
69 isChannelPrivate,
70 filter,
71 bearer,
72 enablePresence,
73 enableDbChanges,
74 enableBroadcast,
75 } = config
76
77 const { data: settings } = useProjectSettingsV2Query({ projectRef: projectRef })
78
79 const protocol = settings?.app_config?.protocol ?? 'https'
80 const endpoint = settings?.app_config?.endpoint
81 // the default host is prod until the correct one comes through an API call.
82 const host = settings ? `${protocol}://${endpoint}` : `https://${projectRef}.supabase.co`
83
84 const realtimeUrl = `${host}/realtime/v1`.replace(/^http/i, 'ws')
85
86 const [logData, dispatch] = useReducer(reducer, [] as LogData[])
87 const pushMessage = (messageType: string, metadata: any) => {
88 dispatch({ type: 'add', payload: { messageType, metadata } })
89 }
90
91 // Instantiate our client with the Realtime server and params to connect with
92 let [client, setClient] = useState<RealtimeClient>()
93 let [channel, setChannel] = useState<RealtimeChannel | undefined>()
94
95 const roleImpersonationState = useRoleImpersonationStateSnapshot()
96
97 useEffect(() => {
98 if (!enabled) {
99 return
100 }
101
102 const options = {
103 vsn: '2.0.0',
104 headers: DEFAULT_HEADERS,
105 params: { apikey: token, log_level: logLevel },
106 }
107 const realtimeClient = new RealtimeClient(realtimeUrl, options)
108
109 if (bearer) {
110 realtimeClient.setAuth(bearer)
111 }
112
113 setClient(realtimeClient)
114 return () => {
115 realtimeClient.disconnect()
116 setClient(undefined)
117 }
118 }, [enabled, bearer, host, logLevel, token])
119
120 useEffect(() => {
121 if (!client) {
122 return
123 }
124 dispatch({ type: 'clear' })
125 const newChannel = client?.channel(channelName, {
126 config: { broadcast: { self: true }, private: isChannelPrivate },
127 })
128 // Hack to confirm Postgres is subscribed
129 // Need to add 'extension' key in the 'payload'
130 newChannel.on('system' as any, {} as any, (payload: any) => {
131 pushMessage('SYSTEM', payload)
132 })
133
134 if (enableBroadcast) {
135 // Listen for all (`*`) `broadcast` messages
136 // The message name can by anything
137 // Match on specific message names to filter for only those types of messages and do something with them
138 newChannel.on('broadcast', { event: '*' }, (payload) => pushMessage('BROADCAST', payload))
139 }
140
141 // Listen for all (`*`) `presence` messages
142 if (enablePresence) {
143 newChannel.on('presence' as any, { event: '*' }, (payload) => {
144 pushMessage('PRESENCE', payload)
145 })
146 }
147
148 if (enableDbChanges) {
149 let postgres_changes_opts: any = {
150 event: '*',
151 schema: schema,
152 table: table,
153 filter: undefined,
154 }
155 if (filter !== '') {
156 postgres_changes_opts.filter = filter
157 }
158 newChannel.on('postgres_changes' as any, postgres_changes_opts, (payload: any) => {
159 let ts = performance.now() + performance.timeOrigin
160 let payload_ts = Date.parse(payload.commit_timestamp)
161 let latency = ts - payload_ts
162 pushMessage('POSTGRES', { ...payload, latency })
163 })
164 }
165
166 // Finally, subscribe to the Channel we just setup
167 newChannel.subscribe(async (status, err) => {
168 if (status === 'SUBSCRIBED') {
169 // Let LiveView know we connected so we can update the button text
170 // pushMessageTo('#conn_info', 'broadcast_subscribed', { host: host })
171
172 const role = roleImpersonationState.role?.role
173 const computedRole =
174 role === undefined
175 ? 'service_role_'
176 : role === 'anon'
177 ? 'anon_role_'
178 : role === 'authenticated'
179 ? 'authenticated_role_'
180 : 'user_name_'
181
182 if (enablePresence) {
183 const name = computedRole + Math.floor(Math.random() * 100)
184 newChannel.send({
185 type: 'presence',
186 event: 'TRACK',
187 payload: { name: name, t: performance.now() },
188 })
189 }
190 } else if (status === 'CHANNEL_ERROR') {
191 if (err?.message) {
192 toast.error(`Failed to connect with the following error: ${err.message}`)
193 } else {
194 toast.error(`Failed to connect. Please check your RLS policies and try again.`)
195 }
196
197 newChannel.unsubscribe()
198 setChannel(undefined)
199 setRealtimeConfig({ ...config, channelName: '', enabled: false })
200 }
201 })
202
203 setChannel(newChannel)
204 return () => {
205 newChannel.unsubscribe()
206 setChannel(undefined)
207 }
208 }, [
209 client,
210 channelName,
211 enableBroadcast,
212 enableDbChanges,
213 enablePresence,
214 filter,
215 host,
216 schema,
217 table,
218 ])
219
220 const sendMessage = useCallback(
221 async (message: string, payload: any, callback: () => void) => {
222 if (channel) {
223 const res = await channel.send({
224 type: 'broadcast',
225 event: message,
226 payload,
227 })
228 if (res === 'error') {
229 toast.error('Failed to broadcast message')
230 } else {
231 toast.success('Successfully broadcasted message')
232 callback()
233 }
234 } else {
235 toast.error('Failed to broadcast message: channel has not been set')
236 }
237 },
238 [channel]
239 )
240 return { logData, sendMessage }
241}