Migrate Forge bridge to @forge/kvs and add /confluence forge reset command

forge/README.md

@@ -121,20 +121,52 @@ notification.
 
 ### Operational notes
 
-- `register` is one-shot. If you need to rotate the shared secret,
-  clear `mm.registered` from Forge storage first (use `forge install
-  --upgrade` after manually wiping the entry), then re-run the
-  Mattermost wizard.
-- Forge storage values are capped at 240 KiB per entry. The bridge
-  drops the inline page body for events that would exceed this; the
-  channel notification still fires but @-mention DMs are skipped for
-  that single oversized event.
+#### Rotating the shared secret
+
+The `register` endpoint is single-use per secret. To rotate without
+re-running the install wizard, use the in-band command from Mattermost:
+
+```
+/confluence forge reset
+```
+
+This calls the bridge's HMAC-authenticated `reset` web trigger, which
+wipes `mm.registered`, `mm.drainSecret`, and any queued events; the
+plugin then generates a new secret and re-registers automatically.
+Polling resumes within ~30s. No `install cloud` rerun required.
+
+If the in-band reset fails because the plugin and bridge have drifted
+out of sync (i.e. the bridge no longer accepts the plugin's HMAC),
+break-glass via the Forge CLI:
+
+```
+forge invoke -f wipeRegistrationFn -e <environment>
+```
+
+This requires Forge developer access to the app. After it returns,
+run `/confluence install cloud` to re-register.
+
+#### Storage backend
+
+The bridge uses Forge `@forge/kvs` (the legacy `@forge/api` `storage`
+module was removed by Atlassian on 2026-06-22). Keys used:
+
+- `mm.registered` — registration flag (boolean)
+- `mm.drainSecret` — shared HMAC secret (Forge secret-type entry)
+- `evt:*` — buffered Confluence events, drained by the Mattermost plugin
+
+#### Other limits
+
+- KVS values are capped at 240 KiB per entry. The bridge drops the
+  inline page body for events that would exceed this; the channel
+  notification still fires but @-mention DMs are skipped for that
+  single oversized event.
 - Forge web trigger throttle is 1000 req/min per app/environment. At
   a 30-second poll cadence that is 2 req/min per tenant, so one bridge
   accommodates ~500 Confluence Cloud tenants before throttling.
-- Forge storage is wiped 28 days after the app is uninstalled. The
-  bridge is a buffer, not a system of record; the Mattermost plugin
-  is the durable side.
+- Forge KVS is wiped 28 days after the app is uninstalled. The bridge
+  is a buffer, not a system of record; the Mattermost plugin is the
+  durable side.
 
 ## Shape
 

forge/manifest.yml

@@ -48,6 +48,11 @@ modules:
       urlFormat: v2
       response:
         type: dynamic
+    - key: reset
+      function: resetFn
+      urlFormat: v2
+      response:
+        type: dynamic
 
   function:
     - key: enqueueFn
@@ -56,8 +61,12 @@ modules:
       handler: index.drain
     - key: registerFn
       handler: index.register
+    - key: resetFn
+      handler: index.reset
     - key: onInstalledFn
       handler: index.onInstalled
+    - key: wipeRegistrationFn
+      handler: index.wipeRegistration
 
 app:
   runtime:

forge/package.json

@@ -12,7 +12,8 @@
         "install:dev": "forge install --upgrade"
     },
     "dependencies": {
-        "@forge/api": "^4.0.0"
+        "@forge/api": "^4.0.0",
+        "@forge/kvs": "^2.0.0"
     },
     "devDependencies": {
         "@types/js-yaml": "^4.0.9",

forge/src/index.ts

@@ -1,4 +1,5 @@
-import api, { route, storage, webTrigger } from '@forge/api';
+import api, { route, webTrigger } from '@forge/api';
+import { kvs, WhereConditions } from '@forge/kvs';
 import { createHmac, timingSafeEqual } from 'crypto';
 
 const QUEUE_PREFIX = 'evt:';
@@ -34,10 +35,10 @@ export const enqueue = async (event: unknown, context: unknown): Promise<void> =
     const safe = enforceStorageLimit(enriched, contentID);
 
     try {
-        await storage.set(key, { event: safe, context, enqueuedAt: Date.now() });
+        await kvs.set(key, { event: safe, context, enqueuedAt: Date.now() });
         console.log(`enqueue: stored key=${key} bodyAttached=${Boolean(safe.content?.body)}`);
     } catch (err) {
-        console.error(`enqueue: storage.set failed key=${key} error=${(err as Error)?.message ?? err}`);
+        console.error(`enqueue: kvs.set failed key=${key} error=${(err as Error)?.message ?? err}`);
         throw err;
     }
 };
@@ -97,7 +98,7 @@ const enrichWithBody = async (evt: ForgeEvent): Promise<ForgeEvent> => {
 // body with the shared secret set via the `register` trigger.
 export const drain = async (req: WebTriggerRequest): Promise<WebTriggerResponse> => {
     console.log('drain: invoked');
-    const secret = (await storage.getSecret(SECRET_KEY)) as string | undefined;
+    const secret = (await kvs.getSecret(SECRET_KEY)) as string | undefined;
     if (!secret) {
         console.log('drain: rejected, bridge not registered');
         return jsonResponse(503, { error: 'bridge not registered; POST credentials to register web trigger first' });
@@ -119,14 +120,14 @@ export const drain = async (req: WebTriggerRequest): Promise<WebTriggerResponse>
 
     if (body.ack?.length) {
         const ackable = body.ack.filter((k) => typeof k === 'string' && k.startsWith(QUEUE_PREFIX));
-        await Promise.all(ackable.map((k) => storage.delete(k)));
+        await Promise.all(ackable.map((k) => kvs.delete(k)));
         console.log(`drain: acked ${ackable.length} keys`);
     }
 
     const limit = clampLimit(body.limit);
-    const results = await storage
+    const results = await kvs
         .query()
-        .where('key', { condition: 'STARTS_WITH', value: QUEUE_PREFIX })
+        .where('key', WhereConditions.beginsWith(QUEUE_PREFIX))
         .limit(limit)
         .getMany();
 
@@ -135,9 +136,70 @@ export const drain = async (req: WebTriggerRequest): Promise<WebTriggerResponse>
     return jsonResponse(200, { events, nextCursor: results.nextCursor ?? null });
 };
 
-// register is a one-shot. Once `mm.registered` is set, further calls are
-// refused. To re-register, an operator must clear the flag via the Forge CLI:
-//   forge install --upgrade   then re-POST to register
+// reset wipes the registration so a fresh secret can be installed. Authenticated
+// via HMAC using the currently-registered secret, so only a caller that already
+// holds the shared secret (i.e. the Mattermost plugin that registered) can use
+// it. Use the `/confluence forge reset` slash command in Mattermost.
+//
+// When secrets have drifted (the plugin lost its copy, or a different MM
+// instance is trying to re-register) this endpoint cannot help — use the
+// `wipeRegistration` break-glass function via `forge invoke` instead.
+export const reset = async (req: WebTriggerRequest): Promise<WebTriggerResponse> => {
+    console.log('reset: invoked');
+    const secret = (await kvs.getSecret(SECRET_KEY)) as string | undefined;
+    if (!secret) {
+        console.log('reset: bridge not registered, nothing to do');
+        return jsonResponse(200, { ok: true, alreadyClear: true });
+    }
+
+    if (!verifySignature(secret, headerValue(req, 'x-mm-signature'), req.body ?? '')) {
+        console.log('reset: rejected, invalid signature');
+        return jsonResponse(403, { error: 'invalid signature' });
+    }
+
+    const queuedDeleted = await wipeAllStorage();
+    console.log(`reset: cleared registration + ${queuedDeleted} queued events`);
+    return jsonResponse(200, { ok: true, queuedDeleted });
+};
+
+// wipeRegistration is the break-glass equivalent of `reset`. Invoke via the
+// Forge CLI when the in-band reset cannot run (drifted secrets, plugin lost
+// its secret, etc.):
+//
+//   forge invoke -f wipeRegistrationFn -e <env>
+//
+// The Forge CLI authenticates the caller (must have developer access to this
+// app), which is the right gate for a break-glass operation.
+export const wipeRegistration = async (): Promise<{ ok: true; queuedDeleted: number }> => {
+    const queuedDeleted = await wipeAllStorage();
+    console.log(`wipeRegistration: cleared registration + ${queuedDeleted} queued events`);
+    return { ok: true, queuedDeleted };
+};
+
+const wipeAllStorage = async (): Promise<number> => {
+    await kvs.delete(REGISTERED_KEY);
+    await kvs.deleteSecret(SECRET_KEY);
+    let cursor: string | undefined;
+    let deleted = 0;
+    do {
+        const q = kvs
+            .query()
+            .where('key', WhereConditions.beginsWith(QUEUE_PREFIX))
+            .limit(100);
+        if (cursor) q.cursor(cursor);
+        const page = await q.getMany();
+        await Promise.all(page.results.map((r) => kvs.delete(r.key)));
+        deleted += page.results.length;
+        cursor = page.nextCursor ?? undefined;
+    } while (cursor);
+    return deleted;
+};
+
+// register accepts the shared secret used to HMAC-sign drain requests. It is
+// idempotent for the same secret (returns 200 with alreadyRegistered:true). A
+// caller presenting a different secret is rejected with 409; the Mattermost
+// plugin should run `/confluence forge reset` to rotate, or fall back to
+// `forge invoke -f wipeRegistrationFn -e <env>` if the in-band path can't auth.
 export const register = async (req: WebTriggerRequest): Promise<WebTriggerResponse> => {
     let payload: { secret?: string };
     try {
@@ -150,19 +212,28 @@ export const register = async (req: WebTriggerRequest): Promise<WebTriggerRespon
         return jsonResponse(400, { error: 'secret must be at least 32 characters' });
     }
 
-    if (await storage.get(REGISTERED_KEY)) {
-        const existing = (await storage.getSecret(SECRET_KEY)) as string | undefined;
+    if (await kvs.get(REGISTERED_KEY)) {
+        const existing = (await kvs.getSecret(SECRET_KEY)) as string | undefined;
         if (existing && secretsMatch(existing, payload.secret)) {
-            return jsonResponse(200, { ok: true, alreadyRegistered: true });
+            return jsonResponse(200, { ok: true, alreadyRegistered: true, urls: await allWebtriggerURLs() });
         }
         return jsonResponse(409, {
-            error: 'already registered with a different shared secret; clear mm.registered and mm.drainSecret from Forge storage to reset',
+            error: 'already registered with a different shared secret; run `/confluence forge reset` in Mattermost to rotate, or `forge invoke -f wipeRegistrationFn -e <env>` to break-glass',
         });
     }
 
-    await storage.setSecret(SECRET_KEY, payload.secret);
-    await storage.set(REGISTERED_KEY, true);
-    return jsonResponse(200, { ok: true });
+    await kvs.setSecret(SECRET_KEY, payload.secret);
+    await kvs.set(REGISTERED_KEY, true);
+    return jsonResponse(200, { ok: true, urls: await allWebtriggerURLs() });
+};
+
+const allWebtriggerURLs = async (): Promise<{ drain: string; register: string; reset: string }> => {
+    const [drain, register, reset] = await Promise.all([
+        webTrigger.getUrl('drain'),
+        webTrigger.getUrl('register'),
+        webTrigger.getUrl('reset'),
+    ]);
+    return { drain, register, reset };
 };
 
 const secretsMatch = (a: string, b: string): boolean => {

server/command.go

@@ -43,7 +43,8 @@ const (
 	sysAdminHelpText = "\n###### For System Administrators:\n" +
 		"Setup Instructions:\n" +
 		"* `/confluence install cloud` - Connect Mattermost to a Confluence Cloud instance.\n" +
-		"* `/confluence install server` - Connect Mattermost to a Confluence Server or Data Center instance.\n"
+		"* `/confluence install server` - Connect Mattermost to a Confluence Server or Data Center instance.\n" +
+		"* `/confluence forge reset` - Rotate the Forge bridge shared secret in-place (Cloud only).\n"
 
 	invalidCommand              = "Invalid command."
 	installOnlySystemAdmin      = "`/confluence install` can only be run by a system administrator."
@@ -73,6 +74,7 @@ var ConfluenceCommandHandler = Handler{
 		"settings/notifications":     executeNotificationsStatus,
 		"settings/notifications/on":  executeNotificationsOn,
 		"settings/notifications/off": executeNotificationsOff,
+		"forge/reset":                executeForgeReset,
 	},
 	defaultHandler: executeConfluenceDefault,
 }
@@ -142,6 +144,13 @@ func getAutoCompleteData() *model.AutocompleteData {
 	settings.AddCommand(notifications)
 	confluence.AddCommand(settings)
 
+	forge := model.NewAutocompleteData("forge", "", "Manage the Confluence Forge bridge (System Admin)")
+	forge.RoleID = model.SystemAdminRoleId
+	reset := model.NewAutocompleteData("reset", "", "Rotate the Forge bridge shared secret")
+	reset.RoleID = model.SystemAdminRoleId
+	forge.AddCommand(reset)
+	confluence.AddCommand(forge)
+
 	return confluence
 }
 

server/config/main.go

@@ -33,6 +33,8 @@ type Configuration struct {
 	IsCloud           bool   `json:"iscloud"`
 	ForgeSharedSecret string `json:"forgesharedsecret"`
 	ForgeDrainURL     string `json:"forgedrainurl"`
+	ForgeResetURL     string `json:"forgereseturl"`
+	ForgeRegisterURL  string `json:"forgeregisterurl"`
 	ForgeInstallURL   string `json:"forgeinstallurl"`
 }
 

server/flow.go

@@ -847,12 +847,25 @@ func (fm *FlowManager) submitForgeBridgeURLs(_ *flow.Flow, submitted map[string]
 		return "", nil, nil, errors.New("Forge Bridge Shared Secret is not set on this plugin; reload the plugin to regenerate it")
 	}
 
-	if err := postForgeRegister(registerURL, cfg.ForgeSharedSecret); err != nil {
+	urls, err := postForgeRegister(registerURL, cfg.ForgeSharedSecret)
+	if err != nil {
 		errorList["register_url"] = err.Error()
 		return "", nil, errorList, nil
 	}
 
 	cfg.ForgeDrainURL = drainURL
+	cfg.ForgeRegisterURL = registerURL
+	if urls != nil {
+		if urls.Reset != "" {
+			cfg.ForgeResetURL = urls.Reset
+		}
+		if urls.Drain != "" {
+			cfg.ForgeDrainURL = urls.Drain
+		}
+		if urls.Register != "" {
+			cfg.ForgeRegisterURL = urls.Register
+		}
+	}
 	cfg.Sanitize()
 	configMap, err := cfg.ToMap()
 	if err != nil {
@@ -885,18 +898,29 @@ func isForgeWebtriggerURL(raw string) bool {
 	return strings.HasPrefix(u.Path, "/public/")
 }
 
-func postForgeRegister(registerURL, secret string) error {
+type ForgeWebtriggerURLs struct {
+	Drain    string `json:"drain"`
+	Register string `json:"register"`
+	Reset    string `json:"reset"`
+}
+
+type forgeRegisterResponse struct {
+	OK   bool                `json:"ok"`
+	URLs ForgeWebtriggerURLs `json:"urls"`
+}
+
+func postForgeRegister(registerURL, secret string) (*ForgeWebtriggerURLs, error) {
 	body, err := json.Marshal(map[string]string{"secret": secret})
 	if err != nil {
-		return errors.Wrap(err, "failed to encode register payload")
+		return nil, errors.Wrap(err, "failed to encode register payload")
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, registerURL, bytes.NewReader(body))
 	if err != nil {
-		return errors.Wrap(err, "failed to build register request")
+		return nil, errors.Wrap(err, "failed to build register request")
 	}
 	req.Header.Set("Content-Type", "application/json")
 
@@ -909,21 +933,24 @@ func postForgeRegister(registerURL, secret string) error {
 	}
 	resp, err := client.Do(req)
 	if err != nil {
-		return errors.Wrap(err, "failed to reach register URL")
+		return nil, errors.Wrap(err, "failed to reach register URL")
 	}
 	defer resp.Body.Close()
 
 	switch resp.StatusCode {
 	case http.StatusOK, http.StatusNoContent:
-		return nil
+		respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		var parsed forgeRegisterResponse
+		_ = json.Unmarshal(respBody, &parsed) // tolerate older bridges with no urls field
+		return &parsed.URLs, nil
 	case http.StatusConflict:
-		return errors.New("Forge bridge is already registered with a different shared secret. Have the Forge admin delete the `mm.registered` and `mm.drainSecret` storage keys (`forge storage delete mm.registered && forge storage delete mm.drainSecret`), then re-run this wizard.")
+		return nil, errors.New("Forge bridge is already registered with a different shared secret. Run `/confluence forge reset` to rotate, or if that fails ask your Forge admin to run `forge invoke -f wipeRegistrationFn -e <env>`, then re-run this wizard.")
 	default:
 		respBody, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
 		snippet := strings.TrimSpace(string(respBody))
 		if snippet == "" {
 			snippet = resp.Status
 		}
-		return errors.Errorf("bridge rejected registration (%d): %s", resp.StatusCode, snippet)
+		return nil, errors.Errorf("bridge rejected registration (%d): %s", resp.StatusCode, snippet)
 	}
 }

server/forge_poller.go

@@ -291,7 +291,7 @@ func (fp *ForgePoller) handleDrainError(err error) {
 
 	switch httpErr.StatusCode {
 	case http.StatusUnauthorized, http.StatusForbidden:
-		fp.alertOnce(alertKeyHMAC, fmt.Sprintf("Confluence Forge bridge rejected drain request (HTTP %d): the shared secret on this plugin does not match the secret stored in the Forge bridge. Re-run `/confluence install cloud` to re-register, or have the Forge admin delete the `mm.registered` and `mm.drainSecret` storage keys and re-run the wizard.", httpErr.StatusCode))
+		fp.alertOnce(alertKeyHMAC, fmt.Sprintf("Confluence Forge bridge rejected drain request (HTTP %d): the shared secret on this plugin does not match the secret stored in the Forge bridge. Run `/confluence forge reset` to rotate the secret in-place. If that also fails (because the plugin and bridge have drifted out of sync), ask a Forge admin to run `forge invoke -f wipeRegistrationFn -e <env>`, then re-run `/confluence install cloud`.", httpErr.StatusCode))
 	case http.StatusNotFound, http.StatusServiceUnavailable:
 		fp.alertOnce(alertKeyNotRegd, fmt.Sprintf("Confluence Forge bridge reports it is not registered (HTTP %d). Forge events are not being delivered. Re-run `/confluence install cloud` to re-register the bridge.", httpErr.StatusCode))
 	default: