At protocol triggers by ismellike · Pull Request #1100 · Lay3rLabs/WAVS

ismellike · 2025-12-05T17:39:12Z

Working locally with events coming (e2e set up but with a simulated trigger)

TODO:

test over on wavs-at-protocol with updated component Use AT protocol triggers wavs-at-protocol#9

+just lint

dakom

There's a lot of at-protocol-specific knowledge that I'm unsure about (e.g. the different string formats for time, etc.), but overall the implementation looks great!

Aside for the request for moving the config var out, just a couple other high-level things:

Needs an e2e test before we can merge
Please pull in changes from wit repo instead of baking in here
More of a question than request for changes, but we shouldn't merge until we land on an answer (raised elsewhere too): I see there's some logic for kinda dynamically filling in some data like record... maybe we should only use the sequence as the EventIdSalt, and then if they have different content it will break consensus rather than be treated as different events?

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs

into at-protocol-triggers

ismellike · 2025-12-08T21:09:06Z

There's a lot of at-protocol-specific knowledge that I'm unsure about (e.g. the different string formats for time, etc.), but overall the implementation looks great!

Aside for the request for moving the config var out, just a couple other high-level things:

Needs an e2e test before we can merge

Please pull in changes from wit repo instead of baking in here

More of a question than request for changes, but we shouldn't merge until we land on an answer (raised elsewhere too): I see there's some logic for kinda dynamically filling in some data like record... maybe we should only use the sequence as the EventIdSalt, and then if they have different content it will break consensus rather than be treated as different events?

1.)
I have an e2e test in already in packages/layer-tests. Did you want something else set up as well?

2.)
Got that done PR here Lay3rLabs/wavs-wasi#94

3.)
Turns out we should ignore timestamp and sequence for event id generation, and everything else should create a unique id. Done here 2d5fc45
ref timestamps not determinsitic ref sequence not deterministic

improve uniqueness for event id

dakom · 2025-12-09T07:47:26Z

I have an e2e test in already in packages/layer-tests. Did you want something else set up as well?

Not really, I missed it in simulated trigger, nice solution!

BUT - it would be better if we can treat it like our chain triggers, i.e. spin up a local server... otherwise it's not really an "end to end" test, i.e. there's no testing that it really works in the wild

I'll open a separate issue for that though, not a blocker for this PR: #1101

dakom

Really nice work! You covered a ton of ground, even remembering things like to remove triggers, comments, etc.

Left a bunch of comments but I think it's close to getting in :)

packages/layer-tests/src/e2e/runner.rs

packages/layer-tests/src/e2e/test_registry.rs

packages/layer-tests/layer-tests.toml

packages/types/src/packet.rs

packages/types/src/service.rs

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs

packages/wavs/src/subsystems/trigger.rs

dakom · 2025-12-09T08:32:45Z

packages/wavs/src/subsystems/trigger.rs

+                                wanted_dids: None,          // Listen to all repos
+                                cursor: None,
+                                compression: false,
+                                max_message_size: self.config.jetstream_max_message_size,


How does this relate to the message truncation (previous comment, to 512 bytes)?

This is jetstream config https://github.com/bluesky-social/jetstream

maxMessageSizeBytes - The maximum size of a payload that this client would like to receive. Zero means no limit, negative values are treated as zero. (Default "0" or empty = no maximum size)

dakom · 2025-12-09T08:42:23Z

packages/wavs/src/subsystems/trigger.rs

+                        for ((collection_pattern, repo_did_filter, action_filter), lookup_ids) in
+                            triggers_by_atproto_lock.iter()
+                        {
+                            // Skip if already matched by exact lookup


wait, so why are we iterating on the same collection if we're just going to exit for hits we just saw?

this could even be a performance problem if there's lots of triggers... comment isn't quite right, we're "skipping" but only skipping the more expensive pattern-match test, the intersection is itself not free

can you please change it to two completely separate collections:

Exact match (strategy 1, fast path)

Pattern match (like strategy 2, slower path, but will only need to iterate over the triggers in this collection)

?

You can do do this in a follow-up PR if you prefer, since it's a performance issue only at scale and will affect this file quite a bit, if you do - just open an issue to track

updated here 030bcda

dakom · 2025-12-09T08:43:59Z

packages/wavs/src/subsystems/trigger.rs

+                StreamTriggers::AtProto { event } => {
+                    // Convert CommitAction to AtProtoAction
+                    let action_enum = match event.action {
+                        streams::atproto_jetstream::CommitAction::Create => AtProtoAction::Create,


Why not just get rid of CommitAction and use AtProtoAction everywhere?

yep! included here 0151331

ueco-jb

Few suggestions:

there is no rate limiting implemented (or is it the base_delay?), I think we should be careful about that
reconnection logic could be refactored into a separate method for better readability

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs

ueco-jb · 2025-12-10T13:27:16Z

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs

+{
+    let stream = async_stream::stream! {
+        let mut reconnect_count = 0;
+        let max_reconnects = 10;


As David's comment, same here - this should be configurable via JetstreamConfig.

ref comment above

ueco-jb · 2025-12-10T13:29:48Z

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs

+                    let delay = std::cmp::min(
+                        base_delay * 2_u32.pow(reconnect_count),
+                        max_delay
+                    ) + Duration::from_millis(rand::random::<u64>() % 1000);


nit:

Suggested change

) + Duration::from_millis(rand::random::<u64>() % 1000);

) + Duration::from_millis(rand::thread_rng().gen_range(0..1000));

IMO more readable plus random::<u64> can generate big numbers, on the other hand my suggestion requires an extra import. 🤷

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs

ueco-jb · 2025-12-10T13:33:41Z

packages/cli/src/args.rs

+    /// Set an ATProto Jetstream event trigger for a workflow
+    SetAtProtocol {
+        /// Collection NSID to filter for (e.g., "app.bsky.feed.post")
+        /// Supports wildcards with prefix matching (e.g., "app.bsky.feed.*")


Suggested change

/// Supports wildcards with prefix matching (e.g., "app.bsky.feed.*")

/// Supports wildcards with suffix matching (e.g., "app.bsky.feed.*")

the function matches at .*.

imo correct as is, because the prefix is what's matched when using wildcard.

also adds comment on max message size (0 = no max size)

dakom

Thanks for fixing up all those comments, LGTM!

ueco-jb · 2025-12-11T00:16:03Z

I also had two suggestions in the base comment. #1100 (review)

ismellike added 15 commits December 2, 2025 08:18

Progress on at triggers

04d9eaa

has_start_atproto_stream

ac593d7

Update events.wit

91b9333

add config

08cecf9

wkg wit fetch

793184d

Update bindings

cf42c3b

+just lint

at proto support in echo data

7514aef

just wasi-build-native

6d75ba8

atproto echo data test

67a8147

atproto-echo

0c2add3

no jetstream compression

cf3d09a

Working e2e

bc701d9

Update atproto_jetstream.rs

c020429

Update atproto_jetstream.rs

958ecae

Merge branch 'main' into at-protocol-triggers

63ea938

dakom requested changes Dec 8, 2025

View reviewed changes

packages/wavs/src/subsystems/trigger/streams/atproto_jetstream.rs Outdated Show resolved Hide resolved

ismellike added 6 commits December 8, 2025 08:29

Remove default jetstream config

7e8fbbc

Merge branch 'at-protocol-triggers' of https://github.com/Lay3rLabs/WAVS

92f0ea3

into at-protocol-triggers

Update atproto_jetstream.rs

644e5a2

at proto action + cli

35376d5

Update trigger.rs

5b693c0

just download-wit at-protocol-triggers

93f9427

ismellike marked this pull request as ready for review December 8, 2025 20:52

ismellike requested a review from ueco-jb as a code owner December 8, 2025 20:52

ismellike added 2 commits December 8, 2025 15:53

Ignore sequence and timestamp from eventIdSalt creation

2d5fc45

Add rev + op_index

4e722c8

improve uniqueness for event id

dakom mentioned this pull request Dec 9, 2025

e2e JetStream/Bluesky test #1101

Open

dakom requested changes Dec 9, 2025

View reviewed changes

ueco-jb reviewed Dec 10, 2025

View reviewed changes

ismellike added 12 commits December 10, 2025 09:12

all tests

c207ae5

Improve at proto eventidsalt comment

4e973c3

Remove redundant TryFrom atprotoaction

f7c633b

Remove jetstream truncation

1d8a56d

Use strongly-typed JetStreamEvent

ba65bc9

just wasi-build-native

d740547

Improve jetstream types

0151331

Improve connection state handling

fe307d8

Split at_proto triggers into exact and pattern

030bcda

Improve jetstream url creation

026f287

also adds comment on max message size (0 = no max size)

Add metrics for jetstream

291258f

bump version

fc3829a

ismellike requested review from dakom and ueco-jb December 10, 2025 20:29

dakom approved these changes Dec 10, 2025

View reviewed changes

ismellike merged commit 838b71a into main Dec 10, 2025
6 of 8 checks passed

ismellike deleted the at-protocol-triggers branch December 10, 2025 23:49

	) + Duration::from_millis(rand::random::<u64>() % 1000);
	) + Duration::from_millis(rand::thread_rng().gen_range(0..1000));

	/// Supports wildcards with prefix matching (e.g., "app.bsky.feed.*")
	/// Supports wildcards with suffix matching (e.g., "app.bsky.feed.*")

Conversation

ismellike commented Dec 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dakom left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ismellike commented Dec 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dakom commented Dec 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dakom left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ueco-jb left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dakom left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

ueco-jb commented Dec 11, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ismellike commented Dec 5, 2025 •

edited

Loading

dakom left a comment •

edited

Loading

ismellike commented Dec 8, 2025 •

edited

Loading

dakom commented Dec 9, 2025 •

edited

Loading

ueco-jb left a comment •

edited

Loading