From 45e7a392005a112ead08733ab200248b692f33f5 Mon Sep 17 00:00:00 2001 From: Natalie Bridgers Date: Fri, 7 Aug 2026 16:09:01 -0500 Subject: [PATCH] Add streamplaceFormat option for Discord webhooks Introduce a `streamplaceFormat` configuration option to allow posting messages as "[Streamplace]" with the sender's handle inline, rather than using the sender's handle as the webhook name. Implement caching for avatar lookups to improve performance and reliability. Signed-off-by: Natalie Bridgers --- .../components/settings/webhook-manager.tsx | 58 +++++- js/docs/src/content/docs/features/webhooks.md | 4 + .../content/docs/lex-reference/openapi.json | 15 ++ .../place-stream-server-createwebhook.md | 28 +-- .../server/place-stream-server-defs.md | 38 ++-- .../place-stream-server-updatewebhook.md | 30 +-- .../place/stream/server/createWebhook.json | 5 + lexicons/place/stream/server/defs.json | 5 + .../place/stream/server/updateWebhook.json | 5 + pkg/integrations/discord/avatars.go | 80 ++++++-- pkg/integrations/discord/avatars_test.go | 186 ++++++++++++++++++ .../discord/discordtypes/discordtypes.go | 11 +- pkg/integrations/discord/send-chat.go | 67 ++++--- pkg/integrations/discord/send-chat_test.go | 68 +++++++ pkg/integrations/webhook/manager.go | 14 +- pkg/placestream/servercreatewebhook.go | 2 + pkg/placestream/serverdefs.go | 2 + pkg/placestream/serverupdatewebhook.go | 2 + pkg/spxrpc/webhook.go | 3 + pkg/statedb/webhook.go | 32 +-- 20 files changed, 557 insertions(+), 98 deletions(-) create mode 100644 pkg/integrations/discord/avatars_test.go create mode 100644 pkg/integrations/discord/send-chat_test.go diff --git a/js/app/components/settings/webhook-manager.tsx b/js/app/components/settings/webhook-manager.tsx index 8e3cfcaf..309d7e42 100644 --- a/js/app/components/settings/webhook-manager.tsx +++ b/js/app/components/settings/webhook-manager.tsx @@ -53,6 +53,7 @@ interface Webhook { url: string; events: string[]; active: boolean; + streamplaceFormat?: boolean; prefix?: string; suffix?: string; rewrite?: Array<{ from: string; to: string }>; @@ -69,6 +70,7 @@ interface WebhookFormData { url: string; events: string[]; active: boolean; + streamplaceFormat: boolean; prefix: string; suffix: string; rewrite: Array<{ from: string; to: string }>; @@ -235,6 +237,7 @@ function WebhookForm({ url: webhook?.url || "", events: webhook?.events || ["livestream"], active: webhook?.active ?? true, + streamplaceFormat: webhook?.streamplaceFormat || false, prefix: webhook?.prefix || "", suffix: webhook?.suffix || "", rewrite: webhook?.rewrite || [{ from: "", to: "" }], @@ -253,6 +256,7 @@ function WebhookForm({ url: webhook.url || "", events: webhook.events || ["livestream"], active: webhook.active ?? true, + streamplaceFormat: webhook.streamplaceFormat || false, prefix: webhook.prefix || "", suffix: webhook.suffix || "", rewrite: webhook.rewrite || [{ from: "", to: "" }], @@ -266,6 +270,7 @@ function WebhookForm({ url: "", events: ["livestream"], active: true, + streamplaceFormat: false, prefix: "", suffix: "", rewrite: [{ from: "", to: "" }], @@ -568,6 +573,45 @@ function WebhookForm({ /> + {/* Streamplace format toggle */} + + + setFormData((prev) => ({ + ...prev, + streamplaceFormat: !prev.streamplaceFormat, + })) + } + > + + {formData.streamplaceFormat && ( + ✓ + )} + + + + Post chat messages as [Streamplace] + + + Shows the sender's handle inline instead of as the webhook name. + Example: **@handle**: message + + + + + {/* Example message text */} + + {formData.streamplaceFormat ? "[Streamplace]" : "@{username}"} + - {formData.prefix} - {"{username}"} - {formData.suffix} + {formData.streamplaceFormat && ( + {"**@handle**"} + )} + {formData.streamplaceFormat ? ": " : ""} + {formData.prefix}message{formData.suffix} @@ -694,6 +743,7 @@ export default function WebhookManager() { url: data.url as any, events: data.events as WebhookEvent[], active: data.active, + streamplaceFormat: data.streamplaceFormat, prefix: data.prefix || undefined, suffix: data.suffix || undefined, rewrite: rewriteRules.length > 0 ? rewriteRules : undefined, @@ -731,6 +781,8 @@ export default function WebhookManager() { url: data.url as any, events: data.events as WebhookEvent[], active: data.active, + // always send so unchecking the box persists + streamplaceFormat: data.streamplaceFormat, prefix: data.prefix || undefined, suffix: data.suffix || undefined, rewrite: rewriteRules.length > 0 ? rewriteRules : undefined, diff --git a/js/docs/src/content/docs/features/webhooks.md b/js/docs/src/content/docs/features/webhooks.md index 9278c738..e0b22a1c 100644 --- a/js/docs/src/content/docs/features/webhooks.md +++ b/js/docs/src/content/docs/features/webhooks.md @@ -38,6 +38,10 @@ We'd recommend also filling out these optional fields: "[Streamplace] "). Will apply to both Chat and Livestream events! - Suffix: A suffix to add to each message sent by this webhook (e.g., "is now live!"). Will apply to both Chat and Livestream events! +- Post chat messages as [Streamplace]: By default, chat messages are posted + under the sender's handle (e.g., as `@handle` with their avatar). Check this + box to post them as `[Streamplace]` instead, with the sender's handle inline + in the message: `**@handle**: message`. - Text replacements: A list of text replacements to apply to chat messages sent by this webhook. Each replacement consists of a "from" string and a "to" string. For example, you could replace all instances of "foo" with "bar". diff --git a/js/docs/src/content/docs/lex-reference/openapi.json b/js/docs/src/content/docs/lex-reference/openapi.json index 09e35d02..f23842a3 100644 --- a/js/docs/src/content/docs/lex-reference/openapi.json +++ b/js/docs/src/content/docs/lex-reference/openapi.json @@ -633,6 +633,11 @@ "description": "Whether this webhook should be active upon creation.", "default": false }, + "streamplaceFormat": { + "type": "boolean", + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name.", + "default": false + }, "prefix": { "type": "string", "description": "Text to prepend to webhook messages.", @@ -1084,6 +1089,11 @@ "type": "boolean", "description": "Whether this webhook should be active." }, + "streamplaceFormat": { + "type": "boolean", + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name.", + "default": false + }, "prefix": { "type": "string", "description": "Text to prepend to webhook messages.", @@ -6739,6 +6749,11 @@ "type": "boolean", "description": "Whether this webhook is currently active." }, + "streamplaceFormat": { + "type": "boolean", + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name.", + "default": false + }, "prefix": { "type": "string", "description": "Text to prepend to webhook messages.", diff --git a/js/docs/src/content/docs/lex-reference/server/place-stream-server-createwebhook.md b/js/docs/src/content/docs/lex-reference/server/place-stream-server-createwebhook.md index 9499268c..fdd59b6a 100644 --- a/js/docs/src/content/docs/lex-reference/server/place-stream-server-createwebhook.md +++ b/js/docs/src/content/docs/lex-reference/server/place-stream-server-createwebhook.md @@ -24,17 +24,18 @@ Create a new webhook for receiving Streamplace events. **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ------------- | ------------------------------------------------------------------------------------------------------ | ----- | ----------------------------------------------------------------------------------------------------- | ---------------- | -| `url` | `string` | ✅ | The webhook URL where events will be sent. | Format: `uri` | -| `events` | Array of `string` | ✅ | The types of events this webhook should receive. | | -| `active` | `boolean` | ❌ | Whether this webhook should be active upon creation. | Default: `false` | -| `prefix` | `string` | ❌ | Text to prepend to webhook messages. | Max Length: 100 | -| `suffix` | `string` | ❌ | Text to append to webhook messages. | Max Length: 100 | -| `rewrite` | Array of [`place.stream.server.defs#rewriteRule`](/lex-reference/place-stream-server-defs#rewriterule) | ❌ | Text replacement rules for webhook messages. | | -| `name` | `string` | ❌ | A user-friendly name for this webhook. | Max Length: 100 | -| `description` | `string` | ❌ | A description of what this webhook is used for. | Max Length: 500 | -| `muteWords` | Array of `string` | ❌ | Words to filter out from chat messages. Messages containing any of these words will not be forwarded. | | +| Name | Type | Req'd | Description | Constraints | +| ------------------- | ------------------------------------------------------------------------------------------------------ | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------- | +| `url` | `string` | ✅ | The webhook URL where events will be sent. | Format: `uri` | +| `events` | Array of `string` | ✅ | The types of events this webhook should receive. | | +| `active` | `boolean` | ❌ | Whether this webhook should be active upon creation. | Default: `false` | +| `streamplaceFormat` | `boolean` | ❌ | Post chat messages as "[Streamplace]" with the sender's handle inline (e.g. "**@handle**: message") instead of using the sender's handle as the webhook name. | Default: `false` | +| `prefix` | `string` | ❌ | Text to prepend to webhook messages. | Max Length: 100 | +| `suffix` | `string` | ❌ | Text to append to webhook messages. | Max Length: 100 | +| `rewrite` | Array of [`place.stream.server.defs#rewriteRule`](/lex-reference/place-stream-server-defs#rewriterule) | ❌ | Text replacement rules for webhook messages. | | +| `name` | `string` | ❌ | A user-friendly name for this webhook. | Max Length: 100 | +| `description` | `string` | ❌ | A description of what this webhook is used for. | Max Length: 500 | +| `muteWords` | Array of `string` | ❌ | Words to filter out from chat messages. Messages containing any of these words will not be forwarded. | | **Output:** @@ -95,6 +96,11 @@ Create a new webhook for receiving Streamplace events. "default": false, "description": "Whether this webhook should be active upon creation." }, + "streamplaceFormat": { + "type": "boolean", + "default": false, + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name." + }, "prefix": { "type": "string", "maxLength": 100, diff --git a/js/docs/src/content/docs/lex-reference/server/place-stream-server-defs.md b/js/docs/src/content/docs/lex-reference/server/place-stream-server-defs.md index 55aa8671..7988efed 100644 --- a/js/docs/src/content/docs/lex-reference/server/place-stream-server-defs.md +++ b/js/docs/src/content/docs/lex-reference/server/place-stream-server-defs.md @@ -17,22 +17,23 @@ A webhook configuration for receiving Streamplace events. **Properties:** -| Name | Type | Req'd | Description | Constraints | -| --------------- | --------------------------------------- | ----- | ----------------------------------------------------------------------------------------------------- | ------------------ | -| `id` | `string` | ✅ | Unique identifier for this webhook. | | -| `url` | `string` | ✅ | The webhook URL where events will be sent. | Format: `uri` | -| `events` | Array of `string` | ✅ | The types of events this webhook should receive. | | -| `active` | `boolean` | ✅ | Whether this webhook is currently active. | | -| `prefix` | `string` | ❌ | Text to prepend to webhook messages. | Max Length: 100 | -| `suffix` | `string` | ❌ | Text to append to webhook messages. | Max Length: 100 | -| `rewrite` | Array of [`#rewriteRule`](#rewriterule) | ❌ | Text replacement rules for webhook messages. | | -| `createdAt` | `string` | ✅ | When this webhook was created. | Format: `datetime` | -| `updatedAt` | `string` | ❌ | When this webhook was last updated. | Format: `datetime` | -| `name` | `string` | ❌ | A user-friendly name for this webhook. | Max Length: 100 | -| `description` | `string` | ❌ | A description of what this webhook is used for. | Max Length: 500 | -| `lastTriggered` | `string` | ❌ | When this webhook was last triggered. | Format: `datetime` | -| `errorCount` | `integer` | ❌ | Number of consecutive errors for this webhook. | | -| `muteWords` | Array of `string` | ❌ | Words to filter out from chat messages. Messages containing any of these words will not be forwarded. | | +| Name | Type | Req'd | Description | Constraints | +| ------------------- | --------------------------------------- | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------ | +| `id` | `string` | ✅ | Unique identifier for this webhook. | | +| `url` | `string` | ✅ | The webhook URL where events will be sent. | Format: `uri` | +| `events` | Array of `string` | ✅ | The types of events this webhook should receive. | | +| `active` | `boolean` | ✅ | Whether this webhook is currently active. | | +| `streamplaceFormat` | `boolean` | ❌ | Post chat messages as "[Streamplace]" with the sender's handle inline (e.g. "**@handle**: message") instead of using the sender's handle as the webhook name. | Default: `false` | +| `prefix` | `string` | ❌ | Text to prepend to webhook messages. | Max Length: 100 | +| `suffix` | `string` | ❌ | Text to append to webhook messages. | Max Length: 100 | +| `rewrite` | Array of [`#rewriteRule`](#rewriterule) | ❌ | Text replacement rules for webhook messages. | | +| `createdAt` | `string` | ✅ | When this webhook was created. | Format: `datetime` | +| `updatedAt` | `string` | ❌ | When this webhook was last updated. | Format: `datetime` | +| `name` | `string` | ❌ | A user-friendly name for this webhook. | Max Length: 100 | +| `description` | `string` | ❌ | A description of what this webhook is used for. | Max Length: 500 | +| `lastTriggered` | `string` | ❌ | When this webhook was last triggered. | Format: `datetime` | +| `errorCount` | `integer` | ❌ | Number of consecutive errors for this webhook. | | +| `muteWords` | Array of `string` | ❌ | Words to filter out from chat messages. Messages containing any of these words will not be forwarded. | | --- @@ -107,6 +108,11 @@ S3 storage configuration for backups. "type": "boolean", "description": "Whether this webhook is currently active." }, + "streamplaceFormat": { + "type": "boolean", + "default": false, + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name." + }, "prefix": { "type": "string", "maxLength": 100, diff --git a/js/docs/src/content/docs/lex-reference/server/place-stream-server-updatewebhook.md b/js/docs/src/content/docs/lex-reference/server/place-stream-server-updatewebhook.md index 95aa729b..1553f83a 100644 --- a/js/docs/src/content/docs/lex-reference/server/place-stream-server-updatewebhook.md +++ b/js/docs/src/content/docs/lex-reference/server/place-stream-server-updatewebhook.md @@ -24,18 +24,19 @@ Update an existing webhook configuration. **Schema Type:** `object` -| Name | Type | Req'd | Description | Constraints | -| ------------- | ------------------------------------------------------------------------------------------------------ | ----- | ----------------------------------------------------------------------------------------------------- | --------------- | -| `id` | `string` | ✅ | The ID of the webhook to update. | | -| `url` | `string` | ❌ | The webhook URL where events will be sent. | Format: `uri` | -| `events` | Array of `string` | ❌ | The types of events this webhook should receive. | | -| `active` | `boolean` | ❌ | Whether this webhook should be active. | | -| `prefix` | `string` | ❌ | Text to prepend to webhook messages. | Max Length: 100 | -| `suffix` | `string` | ❌ | Text to append to webhook messages. | Max Length: 100 | -| `rewrite` | Array of [`place.stream.server.defs#rewriteRule`](/lex-reference/place-stream-server-defs#rewriterule) | ❌ | Text replacement rules for webhook messages. | | -| `name` | `string` | ❌ | A user-friendly name for this webhook. | Max Length: 100 | -| `description` | `string` | ❌ | A description of what this webhook is used for. | Max Length: 500 | -| `muteWords` | Array of `string` | ❌ | Words to filter out from chat messages. Messages containing any of these words will not be forwarded. | | +| Name | Type | Req'd | Description | Constraints | +| ------------------- | ------------------------------------------------------------------------------------------------------ | ----- | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------- | +| `id` | `string` | ✅ | The ID of the webhook to update. | | +| `url` | `string` | ❌ | The webhook URL where events will be sent. | Format: `uri` | +| `events` | Array of `string` | ❌ | The types of events this webhook should receive. | | +| `active` | `boolean` | ❌ | Whether this webhook should be active. | | +| `streamplaceFormat` | `boolean` | ❌ | Post chat messages as "[Streamplace]" with the sender's handle inline (e.g. "**@handle**: message") instead of using the sender's handle as the webhook name. | Default: `false` | +| `prefix` | `string` | ❌ | Text to prepend to webhook messages. | Max Length: 100 | +| `suffix` | `string` | ❌ | Text to append to webhook messages. | Max Length: 100 | +| `rewrite` | Array of [`place.stream.server.defs#rewriteRule`](/lex-reference/place-stream-server-defs#rewriterule) | ❌ | Text replacement rules for webhook messages. | | +| `name` | `string` | ❌ | A user-friendly name for this webhook. | Max Length: 100 | +| `description` | `string` | ❌ | A description of what this webhook is used for. | Max Length: 500 | +| `muteWords` | Array of `string` | ❌ | Words to filter out from chat messages. Messages containing any of these words will not be forwarded. | | **Output:** @@ -100,6 +101,11 @@ Update an existing webhook configuration. "type": "boolean", "description": "Whether this webhook should be active." }, + "streamplaceFormat": { + "type": "boolean", + "default": false, + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name." + }, "prefix": { "type": "string", "maxLength": 100, diff --git a/lexicons/place/stream/server/createWebhook.json b/lexicons/place/stream/server/createWebhook.json index 93052142..ecafbd39 100644 --- a/lexicons/place/stream/server/createWebhook.json +++ b/lexicons/place/stream/server/createWebhook.json @@ -36,6 +36,11 @@ "default": false, "description": "Whether this webhook should be active upon creation." }, + "streamplaceFormat": { + "type": "boolean", + "default": false, + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name." + }, "prefix": { "type": "string", "maxLength": 100, diff --git a/lexicons/place/stream/server/defs.json b/lexicons/place/stream/server/defs.json index 5d5f0edc..9ae02e98 100644 --- a/lexicons/place/stream/server/defs.json +++ b/lexicons/place/stream/server/defs.json @@ -34,6 +34,11 @@ "type": "boolean", "description": "Whether this webhook is currently active." }, + "streamplaceFormat": { + "type": "boolean", + "default": false, + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name." + }, "prefix": { "type": "string", "maxLength": 100, diff --git a/lexicons/place/stream/server/updateWebhook.json b/lexicons/place/stream/server/updateWebhook.json index e5a2a20e..ceb9780c 100644 --- a/lexicons/place/stream/server/updateWebhook.json +++ b/lexicons/place/stream/server/updateWebhook.json @@ -39,6 +39,11 @@ "type": "boolean", "description": "Whether this webhook should be active." }, + "streamplaceFormat": { + "type": "boolean", + "default": false, + "description": "Post chat messages as \"[Streamplace]\" with the sender's handle inline (e.g. \"**@handle**: message\") instead of using the sender's handle as the webhook name." + }, "prefix": { "type": "string", "maxLength": 100, diff --git a/pkg/integrations/discord/avatars.go b/pkg/integrations/discord/avatars.go index 962172f8..090f052e 100644 --- a/pkg/integrations/discord/avatars.go +++ b/pkg/integrations/discord/avatars.go @@ -3,26 +3,43 @@ package discord import ( "context" "sync" + "time" "github.com/bluesky-social/indigo/xrpc" + "golang.org/x/sync/singleflight" "stream.place/streamplace/pkg/appbsky" "stream.place/streamplace/pkg/aqhttp" ) -var avatarCache = make(map[string]string) -var avatarCacheMutex = sync.Mutex{} +// avatarCacheEntry is a cached avatar lookup result. Negative results (no +// avatar, or a fetch error) are cached too, so a chatter without an avatar +// doesn't trigger a network call on every message they send. +type avatarCacheEntry struct { + url string + hasAvatar bool + err error + fetchedAt time.Time +} -// getAvatarURL gets the avatar URL for a Bluesky from the public appview -// pretty ugly. we're going to replace this with indexing bluesky profiles -// at some point. -func GetAvatarURL(ctx context.Context, did string) (string, error) { - avatarCacheMutex.Lock() - defer avatarCacheMutex.Unlock() +// avatarNegativeTTL is how long "no avatar" and fetch-error results stay +// cached before being re-validated. Avatar hits are cached for the process +// lifetime. Variable so tests can shrink it. +var avatarNegativeTTL = time.Minute - if avatar, ok := avatarCache[did]; ok { - return avatar, nil - } +var avatarCache = struct { + sync.RWMutex + m map[string]avatarCacheEntry +}{m: make(map[string]avatarCacheEntry)} +// avatarFetchGroup collapses concurrent cache misses for the same DID into a +// single fetch. +var avatarFetchGroup singleflight.Group + +// fetchAvatarURL fetches the avatar URL for a Bluesky DID from the public +// appview. Variable so tests can stub it. +var fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + // pretty ugly. we're going to replace this with indexing bluesky profiles + // at some point. xrpc := &xrpc.Client{ Host: "https://public.api.bsky.app", Client: &aqhttp.Client, @@ -34,9 +51,48 @@ func GetAvatarURL(ctx context.Context, did string) (string, error) { } if profile.Avatar != nil { - avatarCache[did] = *profile.Avatar return *profile.Avatar, nil } return "", nil } + +// GetAvatarURL gets the avatar URL for a Bluesky DID from the public appview. +// +// Successful avatar URLs are cached for the process lifetime; "no avatar" and +// error results are cached for avatarNegativeTTL so busy chat doesn't hammer +// the appview on every message. Concurrent misses for the same DID collapse +// into a single fetch. The cache lock is only held for map access, never +// across the network call. +func GetAvatarURL(ctx context.Context, did string) (string, error) { + avatarCache.RLock() + entry, ok := avatarCache.m[did] + avatarCache.RUnlock() + if ok { + if entry.err != nil { + if time.Since(entry.fetchedAt) < avatarNegativeTTL { + return "", entry.err + } + } else if entry.hasAvatar || time.Since(entry.fetchedAt) < avatarNegativeTTL { + return entry.url, nil + } + // stale negative entry: fall through to refetch + } + + v, err, _ := avatarFetchGroup.Do(did, func() (any, error) { + url, fetchErr := fetchAvatarURL(ctx, did) + entry := avatarCacheEntry{fetchedAt: time.Now(), err: fetchErr} + if fetchErr == nil { + entry.url = url + entry.hasAvatar = url != "" + } + avatarCache.Lock() + avatarCache.m[did] = entry + avatarCache.Unlock() + return url, fetchErr + }) + if err != nil { + return "", err + } + return v.(string), nil +} diff --git a/pkg/integrations/discord/avatars_test.go b/pkg/integrations/discord/avatars_test.go new file mode 100644 index 00000000..808088e4 --- /dev/null +++ b/pkg/integrations/discord/avatars_test.go @@ -0,0 +1,186 @@ +package discord + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// resetAvatarCache clears the global avatar cache between tests. +func resetAvatarCache() { + avatarCache.Lock() + avatarCache.m = make(map[string]avatarCacheEntry) + avatarCache.Unlock() +} + +func TestGetAvatarURLCachesHits(t *testing.T) { + resetAvatarCache() + defer resetAvatarCache() + + var calls int32 + original := fetchAvatarURL + fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + atomic.AddInt32(&calls, 1) + return "https://example.com/" + did + ".png", nil + } + defer func() { fetchAvatarURL = original }() + + ctx := context.Background() + url, err := GetAvatarURL(ctx, "did:example:alice") + require.NoError(t, err) + require.Equal(t, "https://example.com/did:example:alice.png", url) + + // Second call is served from cache, no refetch. + url, err = GetAvatarURL(ctx, "did:example:alice") + require.NoError(t, err) + require.Equal(t, "https://example.com/did:example:alice.png", url) + require.Equal(t, int32(1), atomic.LoadInt32(&calls)) +} + +func TestGetAvatarURLCachesMisses(t *testing.T) { + resetAvatarCache() + defer resetAvatarCache() + + var calls int32 + original := fetchAvatarURL + fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + atomic.AddInt32(&calls, 1) + return "", nil + } + defer func() { fetchAvatarURL = original }() + + ctx := context.Background() + url, err := GetAvatarURL(ctx, "did:example:noavatar") + require.NoError(t, err) + require.Empty(t, url) + + // The "no avatar" result is cached, so a second message doesn't refetch. + url, err = GetAvatarURL(ctx, "did:example:noavatar") + require.NoError(t, err) + require.Empty(t, url) + require.Equal(t, int32(1), atomic.LoadInt32(&calls)) +} + +func TestGetAvatarURLCachesErrors(t *testing.T) { + resetAvatarCache() + defer resetAvatarCache() + + oldTTL := avatarNegativeTTL + avatarNegativeTTL = time.Hour + defer func() { avatarNegativeTTL = oldTTL }() + + var calls int32 + original := fetchAvatarURL + fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + atomic.AddInt32(&calls, 1) + return "", errors.New("appview down") + } + defer func() { fetchAvatarURL = original }() + + ctx := context.Background() + _, err := GetAvatarURL(ctx, "did:example:errored") + require.Error(t, err) + + // The error is cached within the TTL, so a second message doesn't refetch. + _, err = GetAvatarURL(ctx, "did:example:errored") + require.Error(t, err) + require.Equal(t, int32(1), atomic.LoadInt32(&calls)) +} + +func TestGetAvatarURLRefetchesAfterNegativeTTL(t *testing.T) { + resetAvatarCache() + defer resetAvatarCache() + + oldTTL := avatarNegativeTTL + avatarNegativeTTL = 10 * time.Millisecond + defer func() { avatarNegativeTTL = oldTTL }() + + var calls int32 + original := fetchAvatarURL + fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + atomic.AddInt32(&calls, 1) + return "", nil + } + defer func() { fetchAvatarURL = original }() + + ctx := context.Background() + _, _ = GetAvatarURL(ctx, "did:example:stale") + _, _ = GetAvatarURL(ctx, "did:example:stale") + require.Equal(t, int32(1), atomic.LoadInt32(&calls)) + + time.Sleep(30 * time.Millisecond) + _, _ = GetAvatarURL(ctx, "did:example:stale") + require.Equal(t, int32(2), atomic.LoadInt32(&calls)) +} + +func TestGetAvatarURLAvatarHitsIgnoreTTL(t *testing.T) { + resetAvatarCache() + defer resetAvatarCache() + + oldTTL := avatarNegativeTTL + avatarNegativeTTL = 10 * time.Millisecond + defer func() { avatarNegativeTTL = oldTTL }() + + var calls int32 + original := fetchAvatarURL + fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + atomic.AddInt32(&calls, 1) + return "https://example.com/avatar.png", nil + } + defer func() { fetchAvatarURL = original }() + + ctx := context.Background() + _, _ = GetAvatarURL(ctx, "did:example:hasavatar") + + time.Sleep(30 * time.Millisecond) + url, err := GetAvatarURL(ctx, "did:example:hasavatar") + require.NoError(t, err) + require.Equal(t, "https://example.com/avatar.png", url) + require.Equal(t, int32(1), atomic.LoadInt32(&calls)) +} + +func TestGetAvatarURLSingleflight(t *testing.T) { + resetAvatarCache() + defer resetAvatarCache() + + var calls int32 + release := make(chan struct{}) + original := fetchAvatarURL + fetchAvatarURL = func(ctx context.Context, did string) (string, error) { + atomic.AddInt32(&calls, 1) + <-release + return "https://example.com/avatar.png", nil + } + defer func() { fetchAvatarURL = original }() + + ctx := context.Background() + results := make(chan error, 10) + for i := 0; i < 10; i++ { + go func() { + url, err := GetAvatarURL(ctx, "did:example:concurrent") + if err != nil { + results <- err + return + } + if url != "https://example.com/avatar.png" { + results <- errors.New("unexpected url: " + url) + return + } + results <- nil + }() + } + + // Wait until the single flight is in flight, then release it. + require.Eventually(t, func() bool { return atomic.LoadInt32(&calls) == 1 }, time.Second, time.Millisecond*5) + close(release) + for i := 0; i < 10; i++ { + require.NoError(t, <-results) + } + + // All 10 callers shared one fetch. + require.Equal(t, int32(1), atomic.LoadInt32(&calls)) +} diff --git a/pkg/integrations/discord/discordtypes/discordtypes.go b/pkg/integrations/discord/discordtypes/discordtypes.go index 6600318c..1eb8d2c4 100644 --- a/pkg/integrations/discord/discordtypes/discordtypes.go +++ b/pkg/integrations/discord/discordtypes/discordtypes.go @@ -1,12 +1,11 @@ package discordtypes type Webhook struct { - DID string `json:"did"` - URL string `json:"url"` - Type string `json:"type"` - Rewrite []*WebhookRewrite `json:"rewrite,omitempty"` - Prefix string `json:"prefix,omitempty"` - Suffix string `json:"suffix,omitempty"` + URL string `json:"url"` + Rewrite []*WebhookRewrite `json:"rewrite,omitempty"` + Prefix string `json:"prefix,omitempty"` + Suffix string `json:"suffix,omitempty"` + StreamplaceFormat bool `json:"streamplaceFormat"` } type WebhookRewrite struct { diff --git a/pkg/integrations/discord/send-chat.go b/pkg/integrations/discord/send-chat.go index 2fd08e51..9c2f4f82 100644 --- a/pkg/integrations/discord/send-chat.go +++ b/pkg/integrations/discord/send-chat.go @@ -21,36 +21,25 @@ func SendChat(ctx context.Context, w *discordtypes.Webhook, did string, scm *pla return fmt.Errorf("failed to cast chat message to streamplace chat message") } - avatarURL, err := GetAvatarURL(ctx, did) - if err != nil { - log.Warn(ctx, "failed to get avatar URL", "err", err) - } - - payload := discordtypes.Payload{ - Username: fmt.Sprintf("@%s", scm.Author.Handle), - Content: fmt.Sprintf("%s%s%s", w.Prefix, msg.Text, w.Suffix), + // The sender's avatar only shows in the default format; the streamplace + // format uses the webhook's own avatar, so skip the network fetch. + var avatarURL string + var err error + if !w.StreamplaceFormat { + avatarURL, err = GetAvatarURL(ctx, did) + if err != nil { + log.Warn(ctx, "failed to get avatar URL", "err", err) + } } - if avatarURL != "" { - payload.AvatarURL = avatarURL - } - - // apply default anti-ping rewrites - payload.Content = strings.ReplaceAll(payload.Content, "@here", "@\u200Bhere") - payload.Content = strings.ReplaceAll(payload.Content, "@everyone", "@\u200Beveryone") - // and for <@{userid/roleid}> - payload.Content = strings.ReplaceAll(payload.Content, "<@", "<@\u200B") - // then apply custom rewrites, in case user wants to undo the above or do something else - for _, rewrite := range w.Rewrite { - payload.Content = strings.ReplaceAll(payload.Content, rewrite.From, rewrite.To) - } + payload := buildChatPayload(w, scm.Author.Handle, msg, avatarURL) jsonPayload, err := json.Marshal(payload) if err != nil { return fmt.Errorf("failed to marshal payload: %w", err) } - log.Warn(ctx, "sending chat to discord", "payload", string(jsonPayload), "for_did", w.DID) + log.Debug(ctx, "sending chat to discord", "payload", string(jsonPayload), "for_did", did) req, err := http.NewRequestWithContext(ctx, "POST", w.URL, bytes.NewReader(jsonPayload)) if err != nil { @@ -76,3 +65,37 @@ func SendChat(ctx context.Context, w *discordtypes.Webhook, did string, scm *pla return nil } + +// buildChatPayload builds the Discord webhook payload for a chat message. +// +// By default the message is posted under the chatter's handle ("@handle", +// with their avatar). With StreamplaceFormat set, it is posted as +// "[Streamplace]" using the webhook's own avatar, with the handle inline: +// "**@handle**: text". The avatar URL is only used in the default format. +func buildChatPayload(w *discordtypes.Webhook, handle string, msg *placestream.ChatMessage, avatarURL string) discordtypes.Payload { + payload := discordtypes.Payload{ + Content: fmt.Sprintf("%s%s%s", w.Prefix, msg.Text, w.Suffix), + } + if w.StreamplaceFormat { + payload.Username = "[Streamplace]" + payload.Content = fmt.Sprintf("**@%s**: %s", handle, payload.Content) + } else { + payload.Username = fmt.Sprintf("@%s", handle) + if avatarURL != "" { + payload.AvatarURL = avatarURL + } + } + + // apply default anti-ping rewrites + payload.Content = strings.ReplaceAll(payload.Content, "@here", "@\u200Bhere") + payload.Content = strings.ReplaceAll(payload.Content, "@everyone", "@\u200Beveryone") + // and for <@{userid/roleid}> + payload.Content = strings.ReplaceAll(payload.Content, "<@", "<@\u200B") + + // then apply custom rewrites, in case user wants to undo the above or do something else + for _, rewrite := range w.Rewrite { + payload.Content = strings.ReplaceAll(payload.Content, rewrite.From, rewrite.To) + } + + return payload +} diff --git a/pkg/integrations/discord/send-chat_test.go b/pkg/integrations/discord/send-chat_test.go new file mode 100644 index 00000000..8793a84e --- /dev/null +++ b/pkg/integrations/discord/send-chat_test.go @@ -0,0 +1,68 @@ +package discord + +import ( + "testing" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/integrations/discord/discordtypes" + "stream.place/streamplace/pkg/placestream" +) + +func testPayload(w *discordtypes.Webhook, text, avatarURL string) discordtypes.Payload { + return buildChatPayload(w, "natalie", &placestream.ChatMessage{Text: text}, avatarURL) +} + +func TestBuildChatPayloadDefaultFormat(t *testing.T) { + w := &discordtypes.Webhook{Prefix: "[Streamplace] ", Suffix: "!"} + + payload := testPayload(w, "hello world", "https://example.com/avatar.png") + + require.Equal(t, "@natalie", payload.Username) + require.Equal(t, "[Streamplace] hello world!", payload.Content) + require.Equal(t, "https://example.com/avatar.png", payload.AvatarURL) +} + +func TestBuildChatPayloadDefaultFormatNoAvatar(t *testing.T) { + w := &discordtypes.Webhook{} + + payload := testPayload(w, "hello", "") + + require.Equal(t, "@natalie", payload.Username) + require.Equal(t, "hello", payload.Content) + require.Empty(t, payload.AvatarURL) +} + +func TestBuildChatPayloadStreamplaceFormat(t *testing.T) { + w := &discordtypes.Webhook{StreamplaceFormat: true, Prefix: "p ", Suffix: " s"} + + payload := testPayload(w, "hello world", "https://example.com/avatar.png") + + require.Equal(t, "[Streamplace]", payload.Username) + require.Equal(t, "**@natalie**: p hello world s", payload.Content) + // The avatar is dropped in streamplace format so the webhook's own + // avatar shows instead of a random chatter's. + require.Empty(t, payload.AvatarURL) +} + +func TestBuildChatPayloadAntiPing(t *testing.T) { + w := &discordtypes.Webhook{StreamplaceFormat: true} + + payload := testPayload(w, "hi @everyone and @here and <@1234>", "") + + require.Equal(t, "**@natalie**: hi @\u200Beveryone and @\u200Bhere and <@\u200B1234>", payload.Content) +} + +func TestBuildChatPayloadCustomRewritesApplyInBothFormats(t *testing.T) { + defaultPayload := testPayload(&discordtypes.Webhook{ + Rewrite: []*discordtypes.WebhookRewrite{{From: "foo", To: "bar"}}, + }, "foo baz", "") + require.Equal(t, "@natalie", defaultPayload.Username) + require.Equal(t, "bar baz", defaultPayload.Content) + + streamplacePayload := testPayload(&discordtypes.Webhook{ + StreamplaceFormat: true, + Rewrite: []*discordtypes.WebhookRewrite{{From: "foo", To: "bar"}}, + }, "foo baz", "") + require.Equal(t, "[Streamplace]", streamplacePayload.Username) + require.Equal(t, "**@natalie**: bar baz", streamplacePayload.Content) +} diff --git a/pkg/integrations/webhook/manager.go b/pkg/integrations/webhook/manager.go index 94e7e4ce..670b9d32 100644 --- a/pkg/integrations/webhook/manager.go +++ b/pkg/integrations/webhook/manager.go @@ -72,10 +72,16 @@ func webhookToDiscordWebhook(webhook *placestream.ServerDefs_Webhook) (*discordt suffix = *webhook.Suffix } + var streamplaceFormat bool + if webhook.StreamplaceFormat != nil { + streamplaceFormat = *webhook.StreamplaceFormat + } + return &discordtypes.Webhook{ - URL: webhook.Url, - Prefix: prefix, - Suffix: suffix, - Rewrite: rewriteRules, + URL: webhook.Url, + Prefix: prefix, + Suffix: suffix, + Rewrite: rewriteRules, + StreamplaceFormat: streamplaceFormat, }, nil } diff --git a/pkg/placestream/servercreatewebhook.go b/pkg/placestream/servercreatewebhook.go index 05387cb1..6215f558 100644 --- a/pkg/placestream/servercreatewebhook.go +++ b/pkg/placestream/servercreatewebhook.go @@ -28,6 +28,8 @@ type ServerCreateWebhook_Input struct { Prefix *string `json:"prefix,omitempty"` // rewrite: Text replacement rules for webhook messages. Rewrite []ServerDefs_RewriteRule `json:"rewrite,omitempty"` + // streamplaceFormat: Post chat messages as "[Streamplace]" with the sender's handle inline (e.g. "**@handle**: message") instead of using the sender's handle as the webhook name. + StreamplaceFormat *bool `json:"streamplaceFormat,omitempty"` // suffix: Text to append to webhook messages. Suffix *string `json:"suffix,omitempty"` // url: The webhook URL where events will be sent. diff --git a/pkg/placestream/serverdefs.go b/pkg/placestream/serverdefs.go index b568b938..3dd6d5c8 100644 --- a/pkg/placestream/serverdefs.go +++ b/pkg/placestream/serverdefs.go @@ -94,6 +94,8 @@ type ServerDefs_Webhook struct { Prefix *string `json:"prefix,omitempty"` // rewrite: Text replacement rules for webhook messages. Rewrite []ServerDefs_RewriteRule `json:"rewrite,omitempty"` + // streamplaceFormat: Post chat messages as "[Streamplace]" with the sender's handle inline (e.g. "**@handle**: message") instead of using the sender's handle as the webhook name. + StreamplaceFormat *bool `json:"streamplaceFormat,omitempty"` // suffix: Text to append to webhook messages. Suffix *string `json:"suffix,omitempty"` // updatedAt: When this webhook was last updated. diff --git a/pkg/placestream/serverupdatewebhook.go b/pkg/placestream/serverupdatewebhook.go index d0ff9562..24566fdf 100644 --- a/pkg/placestream/serverupdatewebhook.go +++ b/pkg/placestream/serverupdatewebhook.go @@ -30,6 +30,8 @@ type ServerUpdateWebhook_Input struct { Prefix *string `json:"prefix,omitempty"` // rewrite: Text replacement rules for webhook messages. Rewrite []ServerDefs_RewriteRule `json:"rewrite,omitempty"` + // streamplaceFormat: Post chat messages as "[Streamplace]" with the sender's handle inline (e.g. "**@handle**: message") instead of using the sender's handle as the webhook name. + StreamplaceFormat *bool `json:"streamplaceFormat,omitempty"` // suffix: Text to append to webhook messages. Suffix *string `json:"suffix,omitempty"` // url: The webhook URL where events will be sent. diff --git a/pkg/spxrpc/webhook.go b/pkg/spxrpc/webhook.go index c49e411c..8e209575 100644 --- a/pkg/spxrpc/webhook.go +++ b/pkg/spxrpc/webhook.go @@ -198,6 +198,9 @@ func (s *Server) handlePlaceStreamServerUpdateWebhook(ctx context.Context, input if input.Active != nil { updates["active"] = *input.Active } + if input.StreamplaceFormat != nil { + updates["streamplace_format"] = *input.StreamplaceFormat + } if input.Prefix != nil { updates["prefix"] = *input.Prefix } diff --git a/pkg/statedb/webhook.go b/pkg/statedb/webhook.go index b74df539..1a6e3113 100644 --- a/pkg/statedb/webhook.go +++ b/pkg/statedb/webhook.go @@ -17,18 +17,19 @@ type Webhook struct { UserDID string `gorm:"column:user_did;not null;index"` URL string `gorm:"column:url;not null"` - Events json.RawMessage `gorm:"column:events;type:json"` - Active bool `gorm:"column:active;default:false"` - Prefix string `gorm:"column:prefix"` - Suffix string `gorm:"column:suffix"` - Rewrite json.RawMessage `gorm:"column:rewrite;type:json"` - MuteWords json.RawMessage `gorm:"column:mute_words;type:json"` - Name string `gorm:"column:name"` - Description string `gorm:"column:description"` - CreatedAt time.Time `gorm:"column:created_at"` - UpdatedAt time.Time `gorm:"column:updated_at"` - LastTriggered *time.Time `gorm:"column:last_triggered"` - ErrorCount int `gorm:"column:error_count;default:0"` + Events json.RawMessage `gorm:"column:events;type:json"` + Active bool `gorm:"column:active;default:false"` + StreamplaceFormat bool `gorm:"column:streamplace_format;default:false"` + Prefix string `gorm:"column:prefix"` + Suffix string `gorm:"column:suffix"` + Rewrite json.RawMessage `gorm:"column:rewrite;type:json"` + MuteWords json.RawMessage `gorm:"column:mute_words;type:json"` + Name string `gorm:"column:name"` + Description string `gorm:"column:description"` + CreatedAt time.Time `gorm:"column:created_at"` + UpdatedAt time.Time `gorm:"column:updated_at"` + LastTriggered *time.Time `gorm:"column:last_triggered"` + ErrorCount int `gorm:"column:error_count;default:0"` } func (w *Webhook) TableName() string { @@ -192,6 +193,10 @@ func (w *Webhook) ToLexicon() (placestream.ServerDefs_Webhook, error) { Rewrite: rewriteRules, } + if w.StreamplaceFormat { + webhook.StreamplaceFormat = &w.StreamplaceFormat + } + if w.Prefix != "" { webhook.Prefix = &w.Prefix } @@ -285,6 +290,9 @@ func WebhookFromLexiconInput(input placestream.ServerCreateWebhook_Input, userDI if input.Suffix != nil { webhook.Suffix = *input.Suffix } + if input.StreamplaceFormat != nil { + webhook.StreamplaceFormat = *input.StreamplaceFormat + } if input.Name != nil { webhook.Name = *input.Name } -- 2.51.2