Add support for Pub/Sub topic references in Encore.ts

Author: eandre

docs/ts/primitives/pubsub.md

@@ -202,3 +202,37 @@ async function example() {
 	await cartEvents.publish({shoppingCartID: 2, event: "item_added"});
 }
 ```
+
+## Topic references
+
+Encore uses static analysis to determine which services are accessing each Pub/Sub topic,
+and what operations each service is performing.
+
+That information is used for features such as rendering architecture diagrams, and is used by Encore Cloud to provision infrastructure correctly and configure IAM permissions.
+
+This means `Topic` objects can't be passed around however you like,
+as it makes static analysis impossible in many cases. To simplify your workflow, given these restrictions,
+Encore supports defining a "reference" to a topic that can be passed around any way you want.
+
+### Using topic references
+
+Define a topic reference by calling `topic.ref<DesiredPermissions>()` from within a service, where `DesiredPermissions` is one of the pre-defined permission types defined in the `encore.dev/pubsub` module. 
+
+This means you're effectively pre-declaring the permissions you need, and only the methods that
+are allowed by those permissions are available on the returned reference object.
+
+For example, to get a reference to a topic that can publish messages:
+
+```typescript
+import { Publisher } from "encore.dev/pubsub";
+const ref = cartEvents.ref<Publisher>();
+
+// You can now freely pass around `ref`, and you can use
+// `ref.publish()` just like you would `cartEvents.publish()`.
+```
+
+To ensure Encore still is aware of which permissions each service needs, the call to `topic.ref`
+must be made from within a service, so that Encore knows which service to associate the permissions with.
+
+Currently, the only permission type is `Publisher`, which allows publishing events to the topic.
+We plan to add more permission types in the future.

runtimes/js/encore.dev/pubsub/mod.ts

@@ -4,6 +4,8 @@ export type { TopicConfig, DeliveryGuarantee } from "./topic";
 export { Subscription } from "./subscription";
 export type { SubscriptionConfig } from "./subscription";
 
+export type { TopicPerms, Publisher } from "./refs";
+
 /**
  * Attribute represents a field on a message that should be sent
  * as an attribute in a PubSub message, rather than in the message
@@ -47,8 +49,8 @@ export type Attribute<T extends string | number | boolean> =
 export type AttributesOf<T extends object> = keyof {
   [Key in keyof // for (const Key in T)
   T as Extract<T[Key], allBrandedTypes> extends never //  if (typeof T[Key] !== oneof(allBrandedTypes))
-  ? never // drop the key
-  : Key]: never; // else keep the key
+    ? never // drop the key
+    : Key]: never; // else keep the key
 };
 
 /**

runtimes/js/encore.dev/pubsub/refs.ts

@@ -0,0 +1,7 @@
+export abstract class TopicPerms {
+  private topicPerms(): void {}
+}
+
+export abstract class Publisher<Msg extends object> extends TopicPerms {
+  abstract publish(msg: Msg): Promise<string>;
+}

runtimes/js/encore.dev/pubsub/topic.ts

@@ -1,17 +1,22 @@
 import { getCurrentRequest } from "../internal/reqtrack/mod";
 import type { AttributesOf } from "./mod";
 import * as runtime from "../internal/runtime/mod";
+import { Publisher, TopicPerms } from "./refs";
 
 /**
  * A topic is a resource to which you can publish messages
  * to be delivered to subscribers of that topic.
  */
-export class Topic<Msg extends object> {
+export class Topic<Msg extends object>
+  extends TopicPerms
+  implements Publisher<Msg>
+{
   public readonly name: string;
   public readonly cfg: TopicConfig<Msg>;
   private impl: runtime.PubSubTopic;
 
   constructor(name: string, cfg: TopicConfig<Msg>) {
+    super();
     this.name = name;
     this.cfg = cfg;
     this.impl = runtime.RT.pubsubTopic(name);
@@ -21,6 +26,10 @@ export class Topic<Msg extends object> {
     const source = getCurrentRequest();
     return this.impl.publish(msg, source);
   }
+
+  public ref<P extends TopicPerms>(): P {
+    return this as unknown as P;
+  }
 }
 
 /**

tsparser/src/legacymeta/mod.rs

@@ -10,6 +10,7 @@ use crate::parser::parser::{ParseContext, ParseResult, Service};
 use crate::parser::resourceparser::bind::{Bind, BindKind};
 use crate::parser::resources::apis::{authhandler, gateway};
 use crate::parser::resources::infra::cron::CronJobSchedule;
+use crate::parser::resources::infra::pubsub_topic::TopicOperation;
 use crate::parser::resources::infra::{cron, objects, pubsub_subscription, pubsub_topic, sqldb};
 use crate::parser::resources::Resource;
 use crate::parser::types::validation;
@@ -453,26 +454,28 @@ impl MetaBuilder<'_> {
         let mut bucket_perms = HashMap::new();
         for u in &self.parse.usages {
             match u {
-                Usage::PublishTopic(publish) => {
-                    let svc =
-                        self.service_for_range(&publish.range)
-                            .ok_or(publish.range.parse_err(
-                                "unable to determine which service this 'publish' call is within",
-                            ))?;
-
-                    // Add the publisher if it hasn't already been seen.
-                    let key = (svc.name.clone(), publish.topic.name.clone());
-                    if seen_publishers.insert(key) {
-                        let service_name = svc.name.clone();
-
-                        let idx = topic_by_name
-                            .get(&publish.topic.name)
-                            .ok_or(publish.range.parse_err("could not resolve topic"))?
-                            .to_owned();
-                        let topic = &mut self.data.pubsub_topics[idx];
-                        topic
-                            .publishers
-                            .push(v1::pub_sub_topic::Publisher { service_name });
+                Usage::Topic(access) => {
+                    if access.ops.contains(&TopicOperation::Publish) {
+                        let svc =
+                            self.service_for_range(&access.range)
+                                .ok_or(access.range.parse_err(
+                                    "cannot determine which service is accessing this topic",
+                                ))?;
+
+                        // Add the publisher if it hasn't already been seen.
+                        let key = (svc.name.clone(), access.topic.name.clone());
+                        if seen_publishers.insert(key) {
+                            let service_name = svc.name.clone();
+
+                            let idx = topic_by_name
+                                .get(&access.topic.name)
+                                .ok_or(access.range.parse_err("could not resolve topic"))?
+                                .to_owned();
+                            let topic = &mut self.data.pubsub_topics[idx];
+                            topic
+                                .publishers
+                                .push(v1::pub_sub_topic::Publisher { service_name });
+                        }
                     }
                 }
                 Usage::AccessDatabase(access) => {

tsparser/src/parser/resources/infra/pubsub_topic.rs

@@ -1,3 +1,5 @@
+use std::ops::Deref;
+
 use litparser_derive::LitParser;
 use swc_common::sync::Lrc;
 use swc_ecma_ast as ast;
@@ -13,8 +15,8 @@ use crate::parser::resources::parseutil::{
     ReferenceParser, TrackedNames,
 };
 use crate::parser::resources::Resource;
-use crate::parser::types::Type;
-use crate::parser::usageparser::{ResolveUsageData, Usage, UsageExprKind};
+use crate::parser::types::{Generic, Type};
+use crate::parser::usageparser::{MethodCall, ResolveUsageData, Usage, UsageExprKind};
 use crate::parser::Range;
 use crate::span_err::ErrReporter;
 
@@ -128,18 +130,38 @@ impl ReferenceParser for PubSubTopicDefinition {
 }
 
 #[derive(Debug)]
-pub struct PublishUsage {
+pub struct TopicUsage {
     pub range: Range,
     pub topic: Lrc<Topic>,
+    pub ops: Vec<TopicOperation>,
 }
 
 pub fn resolve_topic_usage(data: &ResolveUsageData, topic: Lrc<Topic>) -> Option<Usage> {
     match &data.expr.kind {
-        UsageExprKind::MethodCall(method) => {
-            if method.method.as_ref() == "publish" {
-                Some(Usage::PublishTopic(PublishUsage {
+        UsageExprKind::MethodCall(call) => {
+            if call.method.as_ref() == "ref" {
+                let Some(type_args) = call.call.type_args.as_deref() else {
+                    call.call
+                        .span
+                        .err("expected a type argument in call to Topic.ref");
+                    return None;
+                };
+
+                let Some(type_arg) = type_args.params.first() else {
+                    call.call
+                        .span
+                        .err("expected a type argument in call to Topic.ref");
+                    return None;
+                };
+
+                return parse_topic_ref(data, topic, call, type_arg);
+            }
+
+            if call.method.as_ref() == "publish" {
+                Some(Usage::Topic(TopicUsage {
                     range: data.expr.range,
                     topic,
+                    ops: vec![TopicOperation::Publish],
                 }))
             } else {
                 None
@@ -155,3 +177,96 @@ pub fn resolve_topic_usage(data: &ResolveUsageData, topic: Lrc<Topic>) -> Option
         }
     }
 }
+
+fn parse_topic_ref(
+    data: &ResolveUsageData,
+    topic: Lrc<Topic>,
+    _call: &MethodCall,
+    type_arg: &ast::TsType,
+) -> Option<Usage> {
+    fn process_type(
+        data: &ResolveUsageData,
+        sp: &swc_common::Span,
+        t: &Type,
+        depth: usize,
+    ) -> Option<Vec<TopicOperation>> {
+        if depth > 10 {
+            // Prevent infinite recursion.
+            return None;
+        }
+
+        match t {
+            Type::Named(named) => {
+                let ops = match named.obj.name.as_deref() {
+                    Some("Publisher") => vec![TopicOperation::Publish],
+                    _ => {
+                        let underlying = data.type_checker.resolve_obj_type(&named.obj);
+                        return process_type(data, sp, &underlying, depth + 1);
+                    }
+                };
+
+                Some(ops)
+            }
+
+            Type::Class(cls) => {
+                let ops = cls
+                    .methods
+                    .iter()
+                    .filter_map(|method| {
+                        let op = match method.as_str() {
+                            "publish" => TopicOperation::Publish,
+                            _ => {
+                                // Ignore other methods.
+                                return None;
+                            }
+                        };
+
+                        Some(op)
+                    })
+                    .collect();
+                Some(ops)
+            }
+
+            Type::Generic(Generic::Intersection(int)) => {
+                let mut result = Vec::new();
+                for t in &[&int.x, &int.y] {
+                    if let Some(ops) = process_type(data, sp, t, depth + 1) {
+                        result.extend(ops);
+                    }
+                }
+
+                if result.is_empty() {
+                    None
+                } else {
+                    Some(result)
+                }
+            }
+
+            _ => {
+                sp.err(&format!("unsupported topic permission type {t:#?}"));
+                None
+            }
+        }
+    }
+
+    let typ = data
+        .type_checker
+        .resolve_type(data.module.clone(), type_arg);
+
+    if let Some(ops) = process_type(data, &typ.span(), typ.deref(), 0) {
+        Some(Usage::Topic(TopicUsage {
+            range: data.expr.range,
+            topic,
+            ops,
+        }))
+    } else {
+        typ.err("no topic permissions found in type argument");
+        None
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TopicOperation {
+    /// Publishing messages to the topic.
+    Publish,
+}

tsparser/src/parser/usageparser/mod.rs

@@ -237,7 +237,7 @@ impl<'a> UsageResolver<'a> {
 #[derive(Debug)]
 pub enum Usage {
     CallEndpoint(apis::api::CallEndpointUsage),
-    PublishTopic(infra::pubsub_topic::PublishUsage),
+    Topic(infra::pubsub_topic::TopicUsage),
     AccessDatabase(infra::sqldb::AccessDatabaseUsage),
     Bucket(infra::objects::BucketUsage),
 }