[Refactor] Enhance stream reading and subscription APIs for consistency #57
Conversation
--- - Standardize cursor configuration with startFrom method across stream and subscription APIs - Restructure persistent subscription settings with distinct CreateSettings and UpdateSettings for improved flexibility
|
Warning Rate limit exceeded@gradyzhuo has exceeded the limit for the number of commits or files that can be reviewed per hour. Please wait 2 minutes and 45 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (5)
""" WalkthroughThis update refactors persistent subscription APIs by removing explicit cursor parameters from method signatures and consolidating cursor handling within more specific nested option structs. It introduces new stream-specific namespaces, updates builder method names for stream positions, and adds new resource and test files. Documentation and test code are updated to match the revised API. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant PersistentSubscriptions
participant AllStream
participant SpecifiedStream
participant GRPCService
Client->>PersistentSubscriptions: createPersistentSubscription(..., configure: closure)
alt All stream
PersistentSubscriptions->>AllStream: AllStream.Create(group, options)
AllStream->>GRPCService: send create request
else Specified stream
PersistentSubscriptions->>SpecifiedStream: SpecifiedStream.Create(streamId, group, options)
SpecifiedStream->>GRPCService: send create request
end
GRPCService-->>Client: Response
Client->>PersistentSubscriptions: subscribePersistentSubscription(..., configure: closure)
alt All stream
PersistentSubscriptions->>AllStream: AllStream.Read(group, options)
AllStream->>GRPCService: start streaming read
else Specified stream
PersistentSubscriptions->>SpecifiedStream: SpecifiedStream.Read(streamId, group, options)
SpecifiedStream->>GRPCService: start streaming read
end
GRPCService-->>Client: Streamed events
Possibly related PRs
Suggested reviewers
Poem
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
Docstrings generation was requested by @gradyzhuo. * #57 (comment) The following files were modified: * `Sources/EventStoreDB/EventStoreDB.swift` * `Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift` * `Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift` * `Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift` * `Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift` * `Sources/KurrentDB/Streams/Streams.swift` * `Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift` * `Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift` * `Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift` * `Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift`
|
Note Generated docstrings for this pull request at #58 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🔭 Outside diff range comments (3)
Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.List.swift (2)
18-26: 🛠️ Refactor suggestionInconsistent with namespace separation objectives.
This implementation maintains optional
streamIdentifierand provides an initializer without a stream identifier, which contradicts the SpecifiedStream namespace purpose. Other SpecifiedStream operations (like Delete) have been refactored to always require a non-optional StreamIdentifier.Consider aligning this with the namespace separation pattern:
- public let streamIdentifier: StreamIdentifier? + public let streamIdentifier: StreamIdentifier - internal init(stream streamIdentifier: StreamIdentifier) { - self.streamIdentifier = streamIdentifier - } - - internal init() { - self.streamIdentifier = nil - } + internal init(stream streamIdentifier: StreamIdentifier) { + self.streamIdentifier = streamIdentifier + }
28-36: 🛠️ Refactor suggestionConditional logic contradicts SpecifiedStream namespace purpose.
The conditional logic for handling both specific streams and "all" streams within the SpecifiedStream namespace is inconsistent with the separation objectives. SpecifiedStream operations should only handle specific streams.
If this needs to handle AllStream operations, consider moving that functionality to
PersistentSubscriptions.AllStream.List:package func requestMessage() throws -> UnderlyingRequest { try .with { - if let streamIdentifier { - $0.options.listForStream.stream = try streamIdentifier.build() - }else{ - $0.options.listForStream.all = .init() - } + $0.options.listForStream.stream = try streamIdentifier.build() } }Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift (1)
79-98:⚠️ Potential issueCritical: Filter settings are overwritten
The filter configuration set on lines 79-83 is completely overwritten by the assignment on line 85. This will cause any filter settings to be lost.
Apply this diff to fix the issue by combining both configurations:
- if let filter { - $0.all.filter = .make(with: filter) - } else { - $0.all.noFilter = .init() - } - - $0.all = .with{ + $0.all = .with{ + if let filter { + $0.filter = .make(with: filter) + } else { + $0.noFilter = .init() + } + switch position { case .start: $0.start = .init() case .end: $0.end = .init() case let .specified(commitPosition, preparePosition): $0.position = .with { $0.commitPosition = commitPosition $0.preparePosition = preparePosition } } }
🧹 Nitpick comments (9)
Tests/GossipTests/Resources/multiple-events.json (1)
1-16: Identical test resource across suites.
This file duplicates the EventStoreDB test fixture verbatim. Consider centralizing shared test data to reduce maintenance overhead.Sources/KurrentDB/KurrentDB.docc/Articles/Getting started.md (2)
2-2: Inconsistent product name: update "EventStoreDB" to "KurrentDB".
The document header still refers to EventStoreDB, whereas the client library is now KurrentDB. Align terminology for consistency.
174-178: Improve trailing‐closure formatting for readability.
Merge the chained builder calls into a single expression block:- let responses = try await client.readStream("some-stream") { - $0 - .startFrom(revision: .end) - .limit(10) - } + let responses = try await client.readStream("some-stream") { + $0.startFrom(revision: .end) + .limit(10) + }Tests/GossipTests/GossipTests.swift (2)
18-20: Consider improving the test implementation.The test method
connection()is misleading as it only creates and prints aSocketAddresswithout testing actual connection functionality. The hardcoded IPv6 address appears to be a placeholder value.Consider either:
- Implementing a proper connection test that matches the method name
- Renaming the method to reflect what it actually tests (e.g.,
socketAddressCreation())- Adding meaningful assertions instead of just printing
-@Test func connection() async throws { +@Test func socketAddressCreation() async throws { let address = try SocketAddress(ipAddress: "fd28:8756:87af:2:488:e1f2:820:e7d1", port: 80) - print(address) + #expect(address.port == 80) + #expect(address.protocol == .inet6)
21-40: Clean up commented code or implement functionality.The extensive commented code suggests planned functionality but makes the file difficult to read and maintain. Consider either implementing these features properly or removing the commented code until ready for implementation.
If this code is experimental or planned for future implementation, consider:
- Moving it to a separate branch or issue
- Adding TODO comments with clear intentions
- Implementing the functionality with proper error handling and assertions
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (1)
16-73: Consider reducing code duplication in builder methods.While the type safety is excellent, there's significant code duplication between the
CreateSettingsandUpdateSettingsextensions. Consider if a shared implementation could be achieved through protocol composition or default implementations.For example, you could create a base protocol with default implementations:
public protocol PersistentSubscriptionsSettingsCommon { associatedtype SettingsType var settings: SettingsType { set get } } extension PersistentSubscriptionsSettingsCommon where SettingsType: PersistentSubscriptionSettingsProtocol { @discardableResult public func resolveLink() -> Self { withCopy { $0.settings.resolveLink = true } } // ... other common methods }This would require both
CreateSettingsandUpdateSettingsto conform to a common protocol with the same property names.Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift (1)
85-86: Remove unnecessary blank lineThere's an extra blank line before the closing brace.
} - }Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift (2)
18-20: Consider explicit visibility modifiers for propertiesThe properties use default (internal) visibility while the struct is public. Consider making them explicitly
public private(set)for consistency with theUpdatestruct.-let streamIdentifier: StreamIdentifier -let group: String -let options: Options +public private(set) var streamIdentifier: StreamIdentifier +public private(set) var group: String +public private(set) var options: Options
74-83: Nested .with blocks could be simplifiedWhile functionally correct, the nested
.withblocks could be combined for better readability.$0.stream = .with{ switch revision { case .start: $0.start = .init() case .end: $0.end = .init() case .specified(let revision): $0.revision = revision } }Consider combining with the parent
.withblock if the API allows it.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (39)
Sources/EventStoreDB/EventStoreDB.swift(8 hunks)Sources/KurrentDB/Core/PersistenSubscription/PersistentSubscription.Settings.swift(2 hunks)Sources/KurrentDB/KurrentDB.docc/Articles/Appending events.md(1 hunks)Sources/KurrentDB/KurrentDB.docc/Articles/Getting started.md(1 hunks)Sources/KurrentDB/KurrentDB.docc/Articles/Reading events.md(8 hunks)Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift(7 hunks)Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift(2 hunks)Sources/KurrentDB/PersistentSubscriptions/Additions/ReadEvent+Additions.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.ReadOptions.swift(0 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.ReadResponse.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.Subscription.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift(5 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift(2 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift(5 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift(4 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/PersistentSubscriptions.Read.swift(0 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.List.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift(4 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift(1 hunks)Sources/KurrentDB/Streams/Streams.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift(1 hunks)Tests/EventStoreDBTests/Resources/ca.crt(1 hunks)Tests/EventStoreDBTests/Resources/multiple-events.json(1 hunks)Tests/GossipTests/GossipTests.swift(1 hunks)Tests/GossipTests/Resources/ca.crt(1 hunks)Tests/GossipTests/Resources/multiple-events.json(1 hunks)Tests/StreamsTests/StreamsTests.swift(5 hunks)
💤 Files with no reviewable changes (2)
- Sources/KurrentDB/PersistentSubscriptions/Usecase/PersistentSubscriptions.Read.swift
- Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.ReadOptions.swift
🧰 Additional context used
🧬 Code Graph Analysis (8)
Sources/KurrentDB/Streams/Streams.swift (1)
Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift (4)
srartFrom(159-172)revision(178-184)backward(148-153)limit(119-124)
Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift (4)
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift (1)
startFrom(68-73)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift (1)
startFrom(61-66)Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift (1)
startFrom(174-179)Sources/GRPCEncapsulates/Buildable.swift (1)
withCopy(14-18)
Tests/StreamsTests/StreamsTests.swift (2)
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift (1)
startFrom(61-66)Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift (2)
startFrom(231-236)filter(208-213)
Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift (5)
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift (1)
startFrom(68-73)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift (1)
startFrom(61-66)Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift (1)
startFrom(231-236)Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift (1)
startFrom(187-192)Sources/GRPCEncapsulates/Buildable.swift (1)
withCopy(14-18)
Sources/KurrentDB/Core/PersistenSubscription/PersistentSubscription.Settings.swift (1)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (16)
resolveLink(17-20)resolveLink(76-79)extraStatistics(22-25)extraStatistics(81-84)messageTimeout(27-30)messageTimeout(86-89)maxRetryCount(32-35)maxRetryCount(91-94)maxSubscriberCount(47-50)maxSubscriberCount(106-109)liveBufferSize(52-55)liveBufferSize(111-114)readBatchSize(57-60)readBatchSize(116-119)historyBufferSize(62-65)historyBufferSize(121-124)
Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift (1)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (16)
resolveLink(17-20)resolveLink(76-79)extraStatistics(22-25)extraStatistics(81-84)maxRetryCount(32-35)maxRetryCount(91-94)maxSubscriberCount(47-50)maxSubscriberCount(106-109)liveBufferSize(52-55)liveBufferSize(111-114)readBatchSize(57-60)readBatchSize(116-119)historyBufferSize(62-65)historyBufferSize(121-124)messageTimeout(27-30)messageTimeout(86-89)
Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift (6)
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift (1)
requestMessage(24-31)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift (2)
requestMessage(34-40)build(71-86)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift (1)
requestMessage(26-33)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.List.swift (1)
requestMessage(28-36)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift (2)
requestMessage(28-34)build(68-78)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift (2)
requestMessage(35-41)build(72-87)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (1)
Sources/GRPCEncapsulates/Buildable.swift (1)
withCopy(14-18)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: Swift 6.1 on ubuntu-latest
- GitHub Check: Swift 6.0 on ubuntu-latest
🔇 Additional comments (66)
Tests/EventStoreDBTests/Resources/multiple-events.json (1)
1-16: Valid JSON fixture for multiple events.
The structure is correct, and the UUIDs and data fields align with the intended test input.Tests/GossipTests/Resources/ca.crt (1)
1-23: PEM-encoded CA certificate looks valid.
The certificate is correctly formatted for TLS trust anchoring in gossip tests.Sources/KurrentDB/KurrentDB.docc/Articles/Appending events.md (1)
140-175: Documentation example updated correctly.
The switch to thestartFrom(revision:)builder closure is consistent with the new API design. The concurrency snippet remains clear and accurate.Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift (1)
232-236: LGTM! Method rename aligns with API standardization.The method rename from
position(from:)tostartFrom(position:)is consistent with the broader refactoring effort to standardize cursor configuration methods across the codebase.Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.ReadResponse.swift (1)
12-12:✅ Verification successful
Verify consistent usage of the new type alias.
The type alias change to use more explicit namespaced types improves clarity and organization.
Ensure that all related code has been updated to use the new underlying service type:
🏁 Script executed:
#!/bin/bash # Description: Verify all references to the old type alias have been updated # Expected: No references to EventStore_Client_PersistentSubscriptions_ReadResp should remain rg "EventStore_Client_PersistentSubscriptions_ReadResp" --type swiftLength of output: 8840
Approved: Consistent alias usage confirmed
I searched for direct references to
EventStore_Client_PersistentSubscriptions_ReadRespand found occurrences only in generated GRPC/Protobuf files. Those usages are expected, since the alias inPersistentSubscriptions.ReadResponse.swiftpoints to the generatedPersistentSubscriptions.UnderlyingService.Method.Read.Output. No manual code still refers directly to the old type. No further action needed.Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift (1)
188-192: LGTM! Method rename enhances API consistency.The method rename from
revision(from:)tostartFrom(revision:)successfully standardizes the cursor configuration API pattern across stream subscription options.Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift (1)
175-179: LGTM! Clean API consistency improvement.The method rename from
position(from:)tostartFrom(position:)aligns well with the PR objective of standardizing cursor configuration across stream and subscription APIs. The parameter name change fromcursortopositionand the updated documentation maintain clarity and consistency.Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.Subscription.swift (1)
23-23:✅ Verification successful
Verify the type alias change doesn't break existing usages.
The type alias has been updated to reflect the new gRPC service structure. Ensure that all existing references to this
Requesttype alias throughout the codebase are compatible with the new underlying type.
🏁 Script executed:
#!/bin/bash # Description: Check for any remaining references to the old type path that might be broken. # Expected: No references to the old PersistentSubscriptions.Read.UnderlyingRequest type. echo "Searching for references to the old type path..." rg "PersistentSubscriptions\.Read\.UnderlyingRequest" --type swift echo "Searching for usage of Request type alias in PersistentSubscriptions.Subscription context..." ast-grep --pattern $'PersistentSubscriptions.Subscription.$_Request$$$'Length of output: 443
No remaining references to the old
UnderlyingRequest—alias update is safe.A project-wide search found zero occurrences of
PersistentSubscriptions.Read.UnderlyingRequest. The new alias inPersistentSubscriptions.Subscriptioncorrectly points toPersistentSubscriptions.UnderlyingService.Method.Read.Input, and no leftover references to the old type remain. No further changes needed.Tests/EventStoreDBTests/Resources/ca.crt (1)
1-24: LGTM! Test certificate properly formatted.The CA certificate is properly formatted in PEM encoding and has a reasonable validity period for testing purposes. Ensure this is indeed a test certificate and not accidentally including a production certificate in the repository.
Tests/StreamsTests/StreamsTests.swift (1)
119-119: Excellent API standardization!The consistent replacement of
.position(from: .end)with.startFrom(position: .end)across all subscription test methods aligns perfectly with the PR objective to standardize cursor configuration APIs. The changes maintain the same functionality while providing a more intuitive and consistent interface.Also applies to: 148-148, 176-176, 204-204, 232-232
Sources/KurrentDB/PersistentSubscriptions/Additions/ReadEvent+Additions.swift (1)
9-9: Good type simplification.The updated parameter type
PersistentSubscriptions.UnderlyingService.Method.Read.Output.ReadEventis cleaner and more consistent than the previous nested generic type. This aligns well with the namespace restructuring efforts mentioned in the PR objectives.Sources/KurrentDB/Core/PersistenSubscription/PersistentSubscription.Settings.swift (2)
9-68: Excellent separation of create and update concerns.Renaming
SettingstoCreateSettingsand providing all required properties with sensible defaults creates a clear contract for subscription creation. This improves API clarity and type safety.
70-102: Well-designed UpdateSettings structure.The
UpdateSettingsstruct with optional properties is perfect for partial updates, allowing clients to modify only specific subscription parameters. The exclusion ofconsumerStrategyis a good design decision since consumer strategies are typically immutable after subscription creation.Sources/KurrentDB/KurrentDB.docc/Articles/Reading events.md (9)
19-22: LGTM! Documentation correctly reflects the new API pattern.The closure-based builder pattern with
.startFrom(revision: .start)properly demonstrates the new standardized cursor configuration approach mentioned in the PR objectives.
53-56: LGTM! Consistent API usage demonstrated.The example correctly shows the new closure-based pattern for reading streams with authentication.
62-67: LGTM! Proper demonstration of revision-specific reading.The example correctly shows how to use
.startFrom(revision: .specified(10))with the limit configuration.
74-78: LGTM! Backwards reading pattern correctly updated.The example properly demonstrates reading backwards with
.startFrom(revision: .end)and.backwards()within the closure.
134-137: LGTM! AllStream reading pattern correctly updated.The example properly demonstrates the new API for reading from the
$allstream using.startFrom(position: .start).
157-161: LGTM! AllStream with link resolution correctly demonstrated.The example shows proper usage of
.startFrom(position: .start)combined with.resolveLinks().
177-180: LGTM! Position-specific reading correctly updated.The example properly shows how to use
.startFrom(position: .specified(commit: 1110, prepare: 1110))for specific positions.
187-190: LGTM! AllStream backwards reading correctly updated.The example properly demonstrates reading the
$allstream backwards with.startFrom(position: .end).
202-205: LGTM! System events handling example correctly updated.The example properly shows the new API pattern for filtering system events while reading from the
$allstream.Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift (3)
11-16: LGTM! Namespace refactoring correctly implemented.The extension scope has been properly narrowed to
PersistentSubscriptions.SpecifiedStream, and the type aliases correctly reference the underlying service types. This aligns with the PR objective of organizing stream-specific operations under dedicated namespaces.
18-24: LGTM! Non-optional streamIdentifier is appropriate for SpecifiedStream.Making
streamIdentifiernon-optional and simplifying the initializer makes sense for the SpecifiedStream namespace, as these operations always target a specific stream.
26-33: LGTM! Simplified request message construction.The removal of conditional logic for handling optional streamIdentifier is correct since SpecifiedStream operations always have a stream identifier. The request message construction is now cleaner and more predictable.
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift (4)
11-16: LGTM! Proper AllStream namespace implementation.The struct is correctly placed under
PersistentSubscriptions.AllStreamnamespace and properly implements theUnaryUnaryprotocol with appropriate type aliases.
18-22: LGTM! Clean group identifier handling.The
groupproperty and initializer are properly implemented for identifying the subscription group.
24-31: LGTM! Correct AllStream request message construction.The request message properly sets
$0.all = .init()which is appropriate for AllStream operations, distinguishing it from SpecifiedStream operations that set a specific stream identifier.
33-38: LGTM! Proper response handling and mapping.The
sendmethod correctly wraps the gRPC client and maps the response toPersistentSubscription.SubscriptionInfo.Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.List.swift (1)
11-15: LGTM! Namespace and type aliases correctly updated.The extension scope has been properly moved to
PersistentSubscriptions.SpecifiedStreamand type aliases correctly reference the underlying service types.Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift (1)
11-35: LGTM! Well-structured implementation following established patterns.The
Liststruct correctly implements theUnaryUnaryprotocol with proper type definitions and follows the consistent pattern used across other persistent subscription operations. The request message construction properly sets thealloption for listing subscriptions across all streams, and the response mapping to an array ofSubscriptionInfois appropriate.Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift (4)
11-11: Namespace reorganization improves API clarity.Moving
GetInfoto thePersistentSubscriptions.SpecifiedStreamnamespace clearly indicates this operation is for specific streams, improving the overall API design and separation of concerns.
13-15: Consistent typealias updates with full qualification.The typealiases are properly updated with the
PersistentSubscriptionsprefix, maintaining consistency with the new namespace structure.
28-32: Simplified request message construction improves maintainability.The removal of conditional logic for optional stream identifiers makes the code cleaner and more predictable, which is appropriate given that
SpecifiedStreamoperations always require a stream identifier.
18-18:✅ Verification successful
Verify impact of making streamIdentifier non-optional.
The change from optional to non-optional
streamIdentifieris a breaking change that aligns with theSpecifiedStreamnamespace semantics. However, ensure all existing callers have been updated to provide a stream identifier.
🏁 Script executed:
#!/bin/bash # Description: Find all usages of GetInfo to verify they provide streamIdentifier # Expected: All usages should provide streamIdentifier parameter echo "Searching for GetInfo usage patterns..." rg -A 5 -B 2 "GetInfo\(" --type swift echo -e "\nSearching for potential old usage patterns..." rg -A 3 "GetInfo.*group.*:" --type swiftLength of output: 3297
All
SpecifiedStream.GetInfoCalls IncludestreamIdentifier
Verified that every invocation inSources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swiftpassesstream: target.identifierwhen constructingSpecifiedStream.GetInfo. The change to a non-optionalstreamIdentifieris fully covered by existing callers—no further updates are needed.Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift (1)
11-41: LGTM! Consistent implementation following established patterns.The
Deletestruct properly implements theUnaryUnaryprotocol with appropriate type definitions. The use ofDiscardedResponsefor the delete operation is correct since delete operations typically don't return meaningful data. The request message construction correctly sets thealloption andgroupName, following the same pattern as other all-stream operations.Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift (2)
11-40: LGTM! Well-structured replay parked implementation.The
ReplayParkedstruct follows the establishedUnaryUnarypattern correctly. The delegation tooptions.build()for request message construction provides good separation of concerns and makes the code more maintainable.
43-77: Excellent options design with fluent API.The
Optionsstruct is well-designed with:
- Clear
StopAtOptionenum for different stopping criteria- Fluent API using
withCopyfor immutable updates- Proper
@discardableResultannotation for method chaining- Correct translation to underlying message format in
build()The switch statement in
build()properly handles bothnoLimitandpositioncases, mapping them to the appropriate underlying message fields.Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift (3)
12-12: Approve type safety improvement.The change from generic
PersistentSubscription.Settingsto specificPersistentSubscription.CreateSettingsimproves type safety and API clarity.
43-43: Approve type safety improvement.The change to
PersistentSubscription.UpdateSettingsaligns with the separation of create and update concerns.
45-89: Approve conditional assignment pattern for update settings.The conditional assignments correctly handle the optional nature of
UpdateSettingsproperties, ensuring only provided values are updated. This is the appropriate pattern for partial update operations.Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift (4)
11-17: Well-structured protocol conformance.The
StreamStreamprotocol implementation properly defines all required type aliases for gRPC streaming operations.
19-27: Clean initialization with appropriate defaults.The initialization correctly sets
streamIdentifiertonilfor all-stream operations and properly assigns the provided parameters.
67-82: Approve builder pattern implementation.The fluent builder methods properly use
withCopyfor immutable updates and provide clear configuration options.
84-96: Approve proper gRPC message building.The
build()method correctly constructs the underlying gRPC message with appropriate option handling for UUID formats.Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (3)
11-14: Approve generic protocol design.The introduction of
associatedtype SettingsTypeimproves type safety and allows the protocol to work with bothCreateSettingsandUpdateSettings.
75-126: Approve type-safe extension for UpdateSettings.The extension correctly provides the same builder methods for
UpdateSettings, maintaining API consistency across both settings types.
129-129: Approve restriction of deprecated methods.Correctly restricting deprecated methods to only
CreateSettingsprevents their use withUpdateSettings, which makes sense for legacy API compatibility.Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift (5)
49-52: Approve namespace organization with nested structs.The introduction of
SpecifiedStreamandAllStreamas nested structs provides excellent namespace organization and improves API discoverability.
79-82: Approve simplified method signature.Removing explicit cursor parameters and using scoped options improves API consistency and reduces parameter coupling. The
AllStream.Createnamespace clearly indicates the operation context.
84-87: Approve consistent update method pattern.The update method follows the same pattern as create, using
AllStream.Update.Optionsfor proper type safety and namespace organization.
124-127: Approve consistent pattern for specified streams.The
SpecifiedStream.Createpattern maintains consistency with the all-stream operations while properly handling the stream identifier parameter.
130-133: Approve specified stream update implementation.The update method correctly passes both
streamIdentifierandgroupto theSpecifiedStream.Updateusecase, maintaining proper context for stream-specific operations.Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift (1)
85-96:❌ Incorrect review comment
Incorrect field assignment for specified stream operation
The
build()method sets$0.all = .init()on line 87, but this is for aSpecifiedStream.Readoperation. This appears incorrect as it should target a specified stream, not all streams.Please verify the correct field name for specified stream read operations. The current implementation might cause reads to target all streams instead of the specified stream. Consider checking the protobuf definition or other specified stream operations for the correct field name.
🏁 Script executed:
#!/bin/bash # Search for other SpecifiedStream operations to understand the correct field usage ast-grep --pattern 'extension PersistentSubscriptions.SpecifiedStream { $$$ }' # Search for protobuf field assignments in other SpecifiedStream operations rg -A 10 "SpecifiedStream.*build.*UnderlyingMessage"Length of output: 137
🏁 Script executed:
#!/bin/bash set -e # Locate all `build()` implementations in PersistentSubscriptions use cases for comparison rg -n "package func build" -A 8 -B 3 Sources/KurrentDB/PersistentSubscriptions/UsecaseLength of output: 14012
🏁 Script executed:
#!/bin/bash set -e # Display the beginning of the Read.swift file to inspect `streamIdentifier` sed -n '1,50p' Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swiftLength of output: 2265
Ignore
.allfield concern in SpecifiedStream.Read.buildThe
.$0.all = .init()line is part of the RPC‐specificOptionsone-of for read/replay operations, not an instruction to read all streams. BothAllStream.ReadandSpecifiedStream.ReplayParkeduse the same.allnested message in theirbuild()methods, and the actual target stream is determined by the separately setstreamIdentifierinrequestMessages(). No change is required here.Likely an incorrect or invalid review comment.
Sources/EventStoreDB/EventStoreDB.swift (7)
10-10: LGTM: Import addition is appropriate.The addition of
GRPCEncapsulatesimport aligns with the refactoring objectives.
133-133: LGTM: Consistent API naming update.The method name change from
position(from:)tostartFrom(position:)improves API consistency across the codebase.
215-215: LGTM: Consistent API naming update.The method name change from
position(from:)tostartFrom(position:)maintains consistency with the API refactoring.
242-242: LGTM: Consistent API naming update.The method name change from
revision(from:)tostartFrom(revision:)aligns with the standardized cursor configuration approach.
284-284: LGTM: Method signatures updated for API consistency.The updated method signatures using nested option types (
SpecifiedStream.Create.OptionsandAllStream.Create.Options) improve API organization and type safety.Also applies to: 291-291
328-328: LGTM: ReadOptions integration in deprecated API.The method signature update to use the new
ReadOptionsstruct maintains backward compatibility while aligning with the refactored API structure.
339-375: LGTM: Well-designed ReadOptions implementation.The
ReadOptionsstruct demonstrates good design principles:
- Follows immutable builder pattern with
withCopy- Provides sensible defaults (bufferSize: 1000, uuidOption: .string)
- Implements proper protocol conformance
- Encapsulates GRPC message building logic
The implementation is consistent with similar option structs in the codebase.
Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift (2)
18-18: LGTM: Systematic API cleanup removes explicit cursor parameters.The removal of
sinceparameters from method signatures and the adoption of more specific nested option types (e.g.,SpecifiedStream.Create.Options,AllStream.Create.Options) successfully consolidates cursor configuration within the options structs. This change aligns perfectly with the PR objectives for API consistency and improved usability.Also applies to: 32-32, 47-47, 62-62, 78-78, 92-92, 167-167, 183-183, 199-199
22-22: LGTM: Method calls updated to rely on options-based configuration.The removal of explicit cursor parameters from
.create()and.update()calls correctly delegates cursor configuration to the options structs, maintaining the new standardized approach to cursor handling.Also applies to: 36-36, 51-51, 66-66, 171-171, 187-187
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift (3)
27-31: 🛠️ Refactor suggestionUpdate documentation to reflect all-stream specific implementation
The documentation mentions "stream selection (all streams or a specific stream)" but this implementation is specifically for all streams under the
AllStreamnamespace.-/// Constructs the underlying gRPC request message for updating a persistent subscription. -/// -/// Builds the request based on the stream selection (all streams or a specific stream) and the provided cursor position or revision. Throws an error if the stream identifier cannot be built. -/// -/// - Throws: An error if building the stream identifier fails. -/// - Returns: The constructed gRPC request message for the update operation. +/// Constructs the underlying gRPC request message for updating a persistent subscription on all streams. +/// +/// Builds the request with the provided group name and options including cursor position. +/// +/// - Returns: The constructed gRPC request message for the update operation.Likely an incorrect or invalid review comment.
49-66: Well-structured options implementationThe separation of
UpdateSettingsfromCreateSettingsand the fluentstartFromAPI provide a clean and consistent interface. The use ofwithCopypattern ensures immutability, which is a best practice for value types in Swift.
69-88: Excellent position handling implementationThe
build()method properly handles allPositionCursorcases with explicit pattern matching, ensuring compile-time safety. The nested initialization syntax is clean and readable.
| package func send(connection: GRPCClient<Transport>, metadata: Metadata, callOptions: CallOptions) async throws -> Responses { | ||
| let responses = AsyncThrowingStream.makeStream(of: Response.self) | ||
|
|
||
| let writer = PersistentSubscriptions.Subscription.Writer() | ||
| let requestMessages = try requestMessages() | ||
| writer.write(messages: requestMessages) | ||
| Task { | ||
| let client = ServiceClient(wrapping: connection) | ||
| try await client.read(metadata: metadata, options: callOptions) { | ||
| try await $0.write(contentsOf: writer.sender) | ||
| } onResponse: { | ||
| for try await message in $0.messages { | ||
| let response = try handle(message: message) | ||
| responses.continuation.yield(response) | ||
| } | ||
| } | ||
| } | ||
| return try await .init(requests: writer, responses: responses.stream) | ||
| } |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Review async streaming implementation for resource management.
The async streaming implementation looks correct, but ensure proper resource cleanup in case of errors or cancellation.
The current implementation doesn't explicitly handle cleanup if the Task throws or is cancelled. Consider wrapping the streaming logic in a defer block or using withTaskCancellationHandler to ensure proper cleanup of the gRPC connection and continuation.
🏁 Script executed:
#!/bin/bash
# Check if there are other streaming implementations in the codebase that handle cleanup
ast-grep --pattern $'Task {
$$$
try await client.$_($$$) {
$$$
} onResponse: {
$$$
}
}'Length of output: 4685
Ensure proper cleanup in async streaming Tasks
The current implementation launches a detached Task for streaming but doesn’t handle errors or cancellation, which can leak the gRPC connection or leave the continuation open. Wrap your streaming logic in a cancellation-aware context and/or add a defer to always close resources:
Suggested pattern:
// Define cleanup once
let cleanup = {
writer.sender.finish() // close request stream
responses.continuation.finish() // close response stream
}
withTaskCancellationHandler(handler: cleanup) {
Task {
defer { cleanup() }
let client = ServiceClient(wrapping: connection)
try await client.read(metadata: metadata, options: callOptions) {
try await $0.write(contentsOf: writer.sender)
} onResponse: {
for try await message in $0.messages {
let response = try handle(message: message)
responses.continuation.yield(response)
}
}
}
}Please apply this to all streaming use cases for consistency:
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift (lines 38–56)
- Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift (lines 45–55)
- Sources/EventStoreDB/PersistentSubscriptions/ReadAnyTarget.swift (lines 54–63)
🤖 Prompt for AI Agents
In
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift
around lines 38 to 56, the async streaming Task does not handle errors or
cancellation, risking resource leaks. To fix this, define a cleanup closure that
finishes the writer's sender and the responses continuation, then wrap the Task
in withTaskCancellationHandler using this cleanup closure. Inside the Task, add
a defer block to call cleanup, ensuring the gRPC connection and continuation are
properly closed on errors or cancellation.
...rsistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift
Outdated
Show resolved
Hide resolved
...rsistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift
Outdated
Show resolved
Hide resolved
...ntSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift
Show resolved
Hide resolved
...rsistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift
Outdated
Show resolved
Hide resolved
| package func requestMessage() throws -> UnderlyingRequest { | ||
| try .with { | ||
| .with { | ||
| $0.options = options.build() | ||
| $0.options.groupName = group | ||
| switch streamSelection { | ||
| case .all(let cursor): | ||
| $0.options.all = .with{ | ||
| switch cursor { | ||
| case .start: | ||
| $0.start = .init() | ||
| case .end: | ||
| $0.end = .init() | ||
| case let .specified(commitPosition, preparePosition): | ||
| $0.position = .with { | ||
| $0.commitPosition = commitPosition | ||
| $0.preparePosition = preparePosition | ||
| } | ||
| } | ||
| } | ||
| case .specified(let identifier, let cursor): | ||
| $0.options.stream = try .with{ | ||
| $0.streamIdentifier = try identifier.build() | ||
| switch cursor { | ||
| case .start: | ||
| $0.start = .init() | ||
| case .end: | ||
| $0.end = .init() | ||
| case .specified(let revision): | ||
| $0.revision = revision | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| $0.options.groupName = group | ||
| } | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Remove unnecessary throws declaration
The requestMessage() method doesn't throw any errors in its implementation, making the throws declaration unnecessary.
-package func requestMessage() throws -> UnderlyingRequest {
+package func requestMessage() -> UnderlyingRequest {📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| package func requestMessage() throws -> UnderlyingRequest { | |
| try .with { | |
| .with { | |
| $0.options = options.build() | |
| $0.options.groupName = group | |
| switch streamSelection { | |
| case .all(let cursor): | |
| $0.options.all = .with{ | |
| switch cursor { | |
| case .start: | |
| $0.start = .init() | |
| case .end: | |
| $0.end = .init() | |
| case let .specified(commitPosition, preparePosition): | |
| $0.position = .with { | |
| $0.commitPosition = commitPosition | |
| $0.preparePosition = preparePosition | |
| } | |
| } | |
| } | |
| case .specified(let identifier, let cursor): | |
| $0.options.stream = try .with{ | |
| $0.streamIdentifier = try identifier.build() | |
| switch cursor { | |
| case .start: | |
| $0.start = .init() | |
| case .end: | |
| $0.end = .init() | |
| case .specified(let revision): | |
| $0.revision = revision | |
| } | |
| } | |
| } | |
| $0.options.groupName = group | |
| } | |
| } | |
| package func requestMessage() -> UnderlyingRequest { | |
| .with { | |
| $0.options = options.build() | |
| $0.options.groupName = group | |
| } | |
| } |
🤖 Prompt for AI Agents
In
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift
around lines 32 to 37, the requestMessage() method is declared with throws but
does not actually throw any errors. Remove the throws keyword from the method
signature to accurately reflect that it does not throw.
Docstrings generation was requested by @gradyzhuo. * #57 (comment) The following files were modified: * `Sources/EventStoreDB/EventStoreDB.swift` * `Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift` * `Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift` * `Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift` * `Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift` * `Sources/KurrentDB/Streams/Streams.swift` * `Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift` * `Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift` * `Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift` * `Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift`
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift (1)
82-90:⚠️ Potential issueFix incorrect message building for SpecifiedStream.
The
build()method sets$0.all = .init()which is incorrect for aSpecifiedStreamoperation. This should configure stream-specific options instead.This is the same issue identified in the previous review. For
SpecifiedStream.ReplayParked, the build method should not set$0.allbut should instead configure the appropriate stream-specific fields.
🧹 Nitpick comments (2)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (1)
16-80: Consider improving documentation consistency.The
UpdateSettingsextension has comprehensive documentation while theCreateSettingsextension has minimal documentation. Consider adding similar detailed documentation to theCreateSettingsmethods for consistency.Also applies to: 82-171
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift (1)
54-54: Fix MARK comment formatting.The MARK comment should follow the standard format with proper spacing.
-//MARK: - Streams +// MARK: - Streams🧰 Tools
🪛 SwiftLint (0.57.0)
[Warning] 54-54: Prefer at least one space after slashes for comments
(comment_spacing)
[Warning] 54-54: MARK comment should be in valid format. e.g. '// MARK: ...' or '// MARK: - ...'
(mark)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (23)
Sources/EventStoreDB/EventStoreDB.swift(12 hunks)Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift(7 hunks)Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift(2 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift(4 hunks)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift(2 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift(4 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift(1 hunks)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift(1 hunks)Sources/KurrentDB/Streams/Streams.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift(1 hunks)Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (18)
- Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift
- Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift
- Sources/KurrentDB/Streams/Streams.swift
- Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift
- Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift
- Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift
- Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift
- Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift
🧰 Additional context used
🧬 Code Graph Analysis (2)
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift (9)
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift (2)
requestMessage(28-36)send(42-47)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift (2)
requestMessage(28-35)send(42-47)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift (4)
requestMessage(36-41)build(82-101)send(43-48)startFrom(69-74)Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift (3)
requestMessage(29-34)build(80-90)send(41-46)Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift (4)
requestMessage(37-43)build(88-103)send(50-55)startFrom(76-81)Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift (5)
requestMessage(29-33)build(139-199)send(35-47)filter(208-213)startFrom(234-239)Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift (2)
create(88-91)create(156-159)Sources/GRPCEncapsulates/Buildable.swift (1)
withCopy(14-18)Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift (2)
make(18-45)make(107-136)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (1)
Sources/GRPCEncapsulates/Buildable.swift (1)
withCopy(14-18)
🪛 SwiftLint (0.57.0)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift
[Warning] 54-54: Prefer at least one space after slashes for comments
(comment_spacing)
[Warning] 54-54: MARK comment should be in valid format. e.g. '// MARK: ...' or '// MARK: - ...'
(mark)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: Swift 6.1 on ubuntu-latest
- GitHub Check: Swift 6.0 on ubuntu-latest
🔇 Additional comments (14)
Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift (1)
12-14: Excellent refactoring with generic associated types.The use of generic associated types to support both
CreateSettingsandUpdateSettingsis a clean architectural improvement that enables type-safe builder patterns for different use cases.Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift (2)
49-52: Excellent API design with clear namespacing.The introduction of
SpecifiedStreamandAllStreamnested structs provides clear namespacing for stream-specific operations and improves the overall API organization.
88-91: Clean method signature refactoring.The updated method signatures that encapsulate cursor/position configuration within the options structs provide a much cleaner and more maintainable API compared to the previous approach with explicit cursor parameters.
Also applies to: 97-100, 156-159, 166-169
Sources/EventStoreDB/EventStoreDB.swift (11)
10-10: LGTM: Import addition aligns with API changes.The addition of
GRPCEncapsulatesimport is appropriate given the refactoring of the persistent subscription APIs.
108-113: LGTM: Documentation improvements enhance clarity.The expanded documentation provides better parameter descriptions and clarifies the behavior of cursor-based reading.
138-138: LGTM: Method call uses correctstartFromsyntax.The method call correctly uses
startFrom(position:)which resolves the previous typo issue mentioned in past reviews.
163-175: LGTM: Enhanced documentation improves API usability.The documentation now clearly explains the cursor behavior and parameter purpose, making the deprecated API more understandable during the transition period.
198-198: LGTM: Consistent use ofstartFrommethod.The method call properly uses
startFrom(revision:)maintaining consistency with the API refactoring pattern.
219-227: LGTM: Subscription API follows consistent pattern.Both the documentation and method call (
startFrom(position:)) demonstrate the consistent API pattern being established across the codebase.Also applies to: 240-240
253-262: LGTM: Stream subscription maintains consistency.The documentation improvements and
startFrom(revision:)method call align with the overall API standardization effort.Also applies to: 276-276
317-331: LGTM: Persistent subscription creation API improvements.The enhanced documentation and updated method signature with specific option types improve API clarity and type safety.
333-346: LGTM: Consistent persistent subscription API pattern.The documentation and method signature updates follow the same pattern as the specified stream variant, maintaining consistency.
377-393: LGTM: Subscription method with improved options.The updated method signature using the new
ReadOptionstype and enhanced documentation improve the API's usability and type safety.
399-446: LGTM: Well-implemented ReadOptions struct.The new
ReadOptionsstruct demonstrates excellent design with:
- Proper encapsulation with private(set) properties
- Immutable builder pattern with returning
Self- Comprehensive documentation for all public methods
- Correct implementation of the
build()method- Appropriate conformance to
EventStoreOptionsprotocolThis implementation aligns perfectly with the PR objectives of enhancing API consistency.
| package func build() -> UnderlyingMessage { | ||
| .with { | ||
| $0.settings = .make(settings: settings) | ||
|
|
||
| if let filter { | ||
| $0.all.filter = .make(with: filter) | ||
| } else { | ||
| $0.all.noFilter = .init() | ||
| } | ||
|
|
||
| $0.all = .with{ | ||
| switch position { | ||
| case .start: | ||
| $0.start = .init() | ||
| case .end: | ||
| $0.end = .init() | ||
| case let .specified(commitPosition, preparePosition): | ||
| $0.position = .with { | ||
| $0.commitPosition = commitPosition | ||
| $0.preparePosition = preparePosition | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
Fix critical logic error in filter handling.
The filter configuration is being overwritten. Lines 96-100 set the filter on $0.all.filter, but line 102 immediately reassigns $0.all with a new configuration, causing the filter to be lost.
Move the filter logic inside the $0.all assignment:
package func build() -> UnderlyingMessage {
.with {
$0.settings = .make(settings: settings)
- if let filter {
- $0.all.filter = .make(with: filter)
- } else {
- $0.all.noFilter = .init()
- }
-
$0.all = .with{
+ if let filter {
+ $0.filter = .make(with: filter)
+ } else {
+ $0.noFilter = .init()
+ }
+
switch position {
case .start:
$0.start = .init()
case .end:
$0.end = .init()
case let .specified(commitPosition, preparePosition):
$0.position = .with {
$0.commitPosition = commitPosition
$0.preparePosition = preparePosition
}
}
}
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| package func build() -> UnderlyingMessage { | |
| .with { | |
| $0.settings = .make(settings: settings) | |
| if let filter { | |
| $0.all.filter = .make(with: filter) | |
| } else { | |
| $0.all.noFilter = .init() | |
| } | |
| $0.all = .with{ | |
| switch position { | |
| case .start: | |
| $0.start = .init() | |
| case .end: | |
| $0.end = .init() | |
| case let .specified(commitPosition, preparePosition): | |
| $0.position = .with { | |
| $0.commitPosition = commitPosition | |
| $0.preparePosition = preparePosition | |
| } | |
| } | |
| } | |
| } | |
| } | |
| package func build() -> UnderlyingMessage { | |
| .with { | |
| $0.settings = .make(settings: settings) | |
| $0.all = .with { | |
| if let filter { | |
| $0.filter = .make(with: filter) | |
| } else { | |
| $0.noFilter = .init() | |
| } | |
| switch position { | |
| case .start: | |
| $0.start = .init() | |
| case .end: | |
| $0.end = .init() | |
| case let .specified(commitPosition, preparePosition): | |
| $0.position = .with { | |
| $0.commitPosition = commitPosition | |
| $0.preparePosition = preparePosition | |
| } | |
| } | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
In
Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift
around lines 92 to 116, the filter is set on $0.all.filter but then $0.all is
immediately reassigned, overwriting the filter setting. To fix this, move the
filter logic inside the $0.all = .with { ... } block so that the filter or
noFilter is set within that assignment, ensuring the filter configuration is not
lost.
…and documentation --- - Fix syntax error in Reading events documentation and update file headers - Remove incorrect all field initialization in ReplayParked options
Docstrings generation was requested by @gradyzhuo. * #57 (comment) The following files were modified: * `Sources/EventStoreDB/EventStoreDB.swift` * `Sources/KurrentDB/KurrentDBClient+PersistentSubscriptions.swift` * `Sources/KurrentDB/PersistentSubscriptions/Additions/EventStore_Client_PersistentSubscriptions+Additions.swift` * `Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.swift` * `Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptionsSettingsBuildable.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Create.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Delete.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.GetInfo.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.List.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.ReplayParked.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Update.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Create.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Delete.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.GetInfo.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.ReplayParked.swift` * `Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Update.swift` * `Sources/KurrentDB/Streams/Streams.swift` * `Sources/KurrentDB/Streams/Usecase/All/Streams.ReadAll.swift` * `Sources/KurrentDB/Streams/Usecase/All/Streams.SubscribeAll.swift` * `Sources/KurrentDB/Streams/Usecase/Specified/Streams.Read.swift` * `Sources/KurrentDB/Streams/Usecase/Specified/Streams.Subscribe.swift`
1 participant
Summary by CodeRabbit
New Features
Bug Fixes
position(from:)andrevision(from:)tostartFrom(position:)andstartFrom(revision:).Refactor
Documentation
Tests