add new functionality and improve backend server structure
- Enhance server functionality and optimize backend code - Add multi-database support (PostgreSQL, MySQL, SQLite)
This commit is contained in:
Tola Leng
@@ -0,0 +1,260 @@
|
||||
package apis
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/pocketbase/dbx"
|
||||
"github.com/pocketbase/pocketbase/core"
|
||||
"github.com/pocketbase/pocketbase/tools/subscriptions"
|
||||
)
|
||||
|
||||
type IBridgedClient interface {
|
||||
subscriptions.Client
|
||||
|
||||
ClientSubscription() *ClientSubscription
|
||||
|
||||
// IsRemoteClient returns true if the client is connected to current pocketbase instance via WebSocket.
|
||||
// It returns false if the client is connected to other pocketbase instances.
|
||||
IsRemoteClient() bool
|
||||
|
||||
// BroadcastGoOffline notifies all other pocketbase instances that the client is offline.
|
||||
BroadcastGoOffline()
|
||||
|
||||
// BroadcastChanges notifies all other pocketbase instances about the client's subscription changes.
|
||||
// It also works as BroadcastGoOnline() when it is called for the first time.
|
||||
BroadcastChanges()
|
||||
|
||||
// ReceiveChanges receives subscription changes from other pocketbase instances.
|
||||
ReceiveChanges(newSubscription *ClientSubscription, authRecordJson string)
|
||||
}
|
||||
|
||||
// ClientSubscription is stored in the _realtimeClients table in PostgreSQL.
|
||||
// Every time a new client connects to the pocketbase server, a new row is created.
|
||||
// And other pocketbase servers are notified via Postgres's Listen/Notify feature.
|
||||
type ClientSubscription struct {
|
||||
ClientId string `db:"pk,clientId" json:"clientId"`
|
||||
ChannelId string `db:"channelId" json:"channelId"`
|
||||
Subscriptions []string `db:"subscriptions" json:"subscriptions"`
|
||||
AuthCollectionRef string `db:"authCollectionRef" json:"authCollectionRef"`
|
||||
AuthRecordRef string `db:"authRecordRef" json:"authRecordRef"`
|
||||
UpdatedByChannelId string `db:"updatedByChannelId" json:"updatedByChannelId"`
|
||||
}
|
||||
|
||||
var _ subscriptions.Client = (*BridgedClient)(nil)
|
||||
var _ IBridgedClient = (*BridgedClient)(nil)
|
||||
|
||||
// BridgedClient defines a remote subscription client.
|
||||
type BridgedClient struct {
|
||||
subscriptions.Client
|
||||
bridge IRealtimeBridge
|
||||
subscription *ClientSubscription
|
||||
}
|
||||
|
||||
// NewBridgedClient
|
||||
// - NewBridgedClient(bridge) creates a local client
|
||||
// - NewBridgedClient(bridge, optionalSubscription) creates a remote client
|
||||
func NewBridgedClient(bridge IRealtimeBridge, optionalSubscription ...*ClientSubscription) *BridgedClient {
|
||||
var client subscriptions.Client
|
||||
var subscription *ClientSubscription
|
||||
|
||||
if len(optionalSubscription) > 0 && optionalSubscription[0] != nil {
|
||||
// remote client (client is connected to other pocketbase instance)
|
||||
client = subscriptions.NewDefaultClient(optionalSubscription[0].ClientId)
|
||||
subscription = optionalSubscription[0]
|
||||
} else {
|
||||
// local client (client is connected to current pocketbase instance)
|
||||
client = subscriptions.NewDefaultClient()
|
||||
subscription = &ClientSubscription{
|
||||
ClientId: client.Id(),
|
||||
ChannelId: bridge.SelfChannelId(),
|
||||
Subscriptions: []string{},
|
||||
AuthCollectionRef: "",
|
||||
AuthRecordRef: "",
|
||||
UpdatedByChannelId: "",
|
||||
}
|
||||
}
|
||||
|
||||
return &BridgedClient{
|
||||
bridge: bridge,
|
||||
Client: client,
|
||||
subscription: subscription,
|
||||
}
|
||||
}
|
||||
|
||||
func (r *BridgedClient) BroadcastGoOffline() {
|
||||
// delete and notify
|
||||
_, err := r.bridge.App().DB().NewQuery(`
|
||||
WITH deleted AS (
|
||||
DELETE FROM "_realtimeClients"
|
||||
WHERE "clientId" = {:clientId}
|
||||
RETURNING *
|
||||
)
|
||||
SELECT pg_notify('shared_bridge_channel', 'subscription_delete|' || deleted."clientId") FROM deleted;
|
||||
`).Bind(dbx.Params{
|
||||
"clientId": r.subscription.ClientId,
|
||||
}).Execute()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "Error deleting subscription:", err)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// BroadcastChanges notifies all other pocketbase instances about the client's subscription changes.
|
||||
// If it is called for the first time, it will act as `BroadcastGoOnline()`.
|
||||
// If the client has an auth record, it will send the auth record JSON to other pocketbase instances.
|
||||
func (r *BridgedClient) BroadcastChanges() {
|
||||
if len(r.RawSubscriptions()) == 0 {
|
||||
// treat no subscriptions as offline
|
||||
r.BroadcastGoOffline()
|
||||
return
|
||||
}
|
||||
|
||||
// 1. clientId should not be changed
|
||||
// 2. channelId should not be changed
|
||||
// 3. update subscriptions
|
||||
r.subscription.Subscriptions = r.RawSubscriptions()
|
||||
// 4. updatedByChannelId is set to the current channelId
|
||||
r.subscription.UpdatedByChannelId = r.bridge.SelfChannelId()
|
||||
|
||||
// 5. update auth collection and record refs
|
||||
var authSQL string
|
||||
var authParams dbx.Params
|
||||
if record, _ := r.Get(RealtimeClientAuthKey).(*core.Record); record != nil {
|
||||
r.subscription.AuthCollectionRef = record.Collection().Id
|
||||
r.subscription.AuthRecordRef = record.Id
|
||||
authSQL = fmt.Sprintf(`SELECT * FROM {{%s}} WHERE id = {:authRecordRef} LIMIT 1`, record.TableName())
|
||||
authParams = dbx.Params{
|
||||
"authRecordRef": record.Id,
|
||||
}
|
||||
} else {
|
||||
r.subscription.AuthCollectionRef = ""
|
||||
r.subscription.AuthRecordRef = ""
|
||||
authSQL = "SELECT WHERE 1=0" // empty rows.
|
||||
}
|
||||
|
||||
// Update using raw SQL query
|
||||
_, err := r.bridge.App().DB().NewQuery(fmt.Sprintf(`
|
||||
WITH
|
||||
updated AS (
|
||||
INSERT INTO "_realtimeClients" (
|
||||
"clientId", "channelId", "subscriptions", "authCollectionRef", "authRecordRef", "updatedByChannelId"
|
||||
) VALUES (
|
||||
{:clientId}, {:channelId}, {:subscriptions}, {:authCollectionRef}, {:authRecordRef}, {:updatedByChannelId}
|
||||
) ON CONFLICT ("clientId") DO UPDATE
|
||||
SET "subscriptions" = EXCLUDED."subscriptions",
|
||||
"authCollectionRef" = EXCLUDED."authCollectionRef",
|
||||
"authRecordRef" = EXCLUDED."authRecordRef",
|
||||
"updatedByChannelId" = EXCLUDED."updatedByChannelId"
|
||||
RETURNING *
|
||||
),
|
||||
auth AS (
|
||||
%s
|
||||
)
|
||||
SELECT pg_notify(
|
||||
'shared_bridge_channel',
|
||||
CONCAT_WS(
|
||||
'|',
|
||||
'subscription_upsert',
|
||||
(SELECT row_to_json(u) FROM updated u)::text,
|
||||
COALESCE((SELECT row_to_json(a) FROM auth a)::text, '')
|
||||
)
|
||||
);
|
||||
`, authSQL)).Bind(authParams).Bind(dbx.Params{
|
||||
"clientId": r.subscription.ClientId,
|
||||
"channelId": r.subscription.ChannelId,
|
||||
"subscriptions": r.subscription.Subscriptions,
|
||||
"authCollectionRef": r.subscription.AuthCollectionRef,
|
||||
"authRecordRef": r.subscription.AuthRecordRef,
|
||||
"updatedByChannelId": r.subscription.UpdatedByChannelId,
|
||||
}).Execute()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "Error updating subscription:", err)
|
||||
return
|
||||
}
|
||||
// other sibling servers will be notified by Postgres Listen/Notify feature
|
||||
}
|
||||
|
||||
func (r *BridgedClient) ReceiveChanges(newSubscription *ClientSubscription, authRecordJson string) {
|
||||
// 1. clientId should not be changed
|
||||
// 2. channelId should not be changed
|
||||
// 3. update subscriptions
|
||||
r.Client.Unsubscribe() // clear previous subscriptions
|
||||
r.Client.Subscribe(newSubscription.Subscriptions...)
|
||||
// 4. update auth collection and record refs
|
||||
if newSubscription.AuthRecordRef == "" || newSubscription.AuthCollectionRef == "" {
|
||||
r.Client.Unset(RealtimeClientAuthKey)
|
||||
} else {
|
||||
authRecord, err := AuthRecordFromJson(r.bridge.App(), newSubscription.AuthCollectionRef, authRecordJson)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, "Error unmarshalling auth record JSON:", err)
|
||||
return
|
||||
}
|
||||
r.Client.Set(RealtimeClientAuthKey, authRecord)
|
||||
}
|
||||
// 5. updatedByChannelId should not be changed
|
||||
|
||||
// Finally, update the subscription
|
||||
r.subscription = newSubscription
|
||||
}
|
||||
|
||||
// Send sends the specified message to the client's channel (if not discarded).
|
||||
func (r *BridgedClient) Send(m subscriptions.Message) {
|
||||
if r.IsDiscarded() {
|
||||
return
|
||||
}
|
||||
|
||||
if r.IsRemoteClient() {
|
||||
// send to sibling servers
|
||||
r.bridge.SendViaBridge(r.subscription.ChannelId, r.Id(), m)
|
||||
} else {
|
||||
// send to connected clients
|
||||
r.Client.Send(m)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *BridgedClient) IsRemoteClient() bool {
|
||||
return r.subscription.ChannelId != r.bridge.SelfChannelId()
|
||||
}
|
||||
|
||||
func (r *BridgedClient) ClientSubscription() *ClientSubscription {
|
||||
return r.subscription
|
||||
}
|
||||
|
||||
func AuthRecordFromJson(app core.App, authCollectionNameOrId string, authRecordJson string) (*core.Record, error) {
|
||||
if authRecordJson == "" {
|
||||
return nil, nil
|
||||
}
|
||||
collection, err := app.FindCachedCollectionByNameOrId(authCollectionNameOrId)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error finding collection by name or ID: %w", err)
|
||||
}
|
||||
authRecord := core.NewRecord(collection)
|
||||
|
||||
data := make(map[string]any)
|
||||
_ = json.Unmarshal([]byte(authRecordJson), &data)
|
||||
for _, field := range collection.Fields {
|
||||
fieldName := field.GetName()
|
||||
value, _ := field.PrepareValue(authRecord, data[fieldName])
|
||||
authRecord.SetRaw(fieldName, value)
|
||||
}
|
||||
return authRecord, nil
|
||||
}
|
||||
|
||||
func split2(s string, sep string) (string, string, bool) {
|
||||
parts := strings.SplitN(s, sep, 2)
|
||||
if len(parts) != 2 {
|
||||
return "", "", false
|
||||
}
|
||||
return parts[0], parts[1], true
|
||||
}
|
||||
|
||||
func split3(s string, sep string) (string, string, string, bool) {
|
||||
parts := strings.SplitN(s, sep, 3)
|
||||
if len(parts) != 3 {
|
||||
return "", "", "", false
|
||||
}
|
||||
return parts[0], parts[1], parts[2], true
|
||||
}
|
||||
Reference in New Issue
Block a user