krz/domain-dig
an ios app for DNS & SSL analysis
clone: git clone https://gitbay.org/krz/domain-dig.git
v5.0.1: DomainDig/IntegrationService.swift · raw
1import Foundation
2import Network
3import os
4import Observation
5import Security
6
7@MainActor
8@Observable
9final class IntegrationService {
10 static let shared = IntegrationService()
11
12 var targets: [IntegrationTarget]
13 var deliveryRecords: [DeliveryRecord]
14 var queue: [QueuedDelivery]
15 var statusMessage: String?
16
17 private let defaults: UserDefaults
18 private var processingTask: Task<Void, Never>?
19
20 private init(defaults: UserDefaults = .standard) {
21 self.defaults = defaults
22 self.targets = Self.loadTargets(defaults: defaults)
23 self.deliveryRecords = Self.loadRecords(defaults: defaults)
24 self.queue = Self.loadQueue(defaults: defaults)
25 }
26
27 func refresh() {
28 targets = Self.loadTargets(defaults: defaults)
29 deliveryRecords = Self.loadRecords(defaults: defaults)
30 queue = Self.loadQueue(defaults: defaults)
31 }
32
33 func upsert(
34 target: IntegrationTarget,
35 webhookURL: String? = nil,
36 slackWebhookURL: String? = nil,
37 emailPassword: String? = nil
38 ) throws {
39 var updatedTarget = target
40
41 switch updatedTarget.configuration {
42 case .webhook(var configuration):
43 if let webhookURL {
44 try Self.validateHTTPS(webhookURL)
45 let reference = configuration.credentialReference ?? Self.secretReference(for: updatedTarget.id, suffix: "webhook")
46 try IntegrationSecretStore.save(secret: webhookURL, reference: reference)
47 configuration.credentialReference = reference
48 configuration.endpointDisplayHost = Self.hostLabel(from: webhookURL)
49 updatedTarget.configuration = .webhook(configuration)
50 }
51 case .slack(var configuration):
52 if let slackWebhookURL {
53 try Self.validateHTTPS(slackWebhookURL)
54 let reference = configuration.credentialReference ?? Self.secretReference(for: updatedTarget.id, suffix: "slack")
55 try IntegrationSecretStore.save(secret: slackWebhookURL, reference: reference)
56 configuration.credentialReference = reference
57 configuration.destinationLabel = Self.hostLabel(from: slackWebhookURL)
58 updatedTarget.configuration = .slack(configuration)
59 }
60 case .email(var configuration):
61 if let emailPassword {
62 let reference = configuration.credentialReference ?? Self.secretReference(for: updatedTarget.id, suffix: "smtp")
63 try IntegrationSecretStore.save(secret: emailPassword, reference: reference)
64 configuration.credentialReference = reference
65 updatedTarget.configuration = .email(configuration)
66 }
67 }
68
69 if let index = targets.firstIndex(where: { $0.id == updatedTarget.id }) {
70 targets[index] = updatedTarget
71 } else {
72 targets.append(updatedTarget)
73 }
74 targets.sort { $0.name.localizedCaseInsensitiveCompare($1.name) == .orderedAscending }
75 persistTargets()
76 statusMessage = "Saved integration settings."
77 }
78
79 func delete(targetID: UUID) {
80 guard let target = targets.first(where: { $0.id == targetID }) else { return }
81 deleteSecrets(for: target)
82 targets.removeAll { $0.id == targetID }
83 queue.removeAll { $0.integrationID == targetID }
84 deliveryRecords.removeAll { $0.integrationID == targetID }
85 persistTargets()
86 persistQueue()
87 persistRecords()
88 }
89
90 func setEnabled(_ isEnabled: Bool, for targetID: UUID) {
91 guard let index = targets.firstIndex(where: { $0.id == targetID }) else { return }
92 targets[index].isEnabled = isEnabled
93 persistTargets()
94 }
95
96 func deliveryRecords(for targetID: UUID) -> [DeliveryRecord] {
97 deliveryRecords
98 .filter { $0.integrationID == targetID }
99 .sorted { $0.timestamp > $1.timestamp }
100 }
101
102 func enqueue(events: [MonitoringEvent]) {
103 guard !events.isEmpty else { return }
104 for event in events {
105 for target in targets {
106 guard target.isEnabled else {
107 appendRecord(
108 DeliveryRecord(
109 integrationID: target.id,
110 eventID: event.id,
111 status: .skipped,
112 destination: destinationLabel(for: target),
113 summary: event.summary,
114 failureReason: Self.disabledTargetReason
115 )
116 )
117 continue
118 }
119
120 if let reason = filterMismatchReason(for: event, target: target) {
121 appendRecord(
122 DeliveryRecord(
123 integrationID: target.id,
124 eventID: event.id,
125 status: .skipped,
126 destination: destinationLabel(for: target),
127 summary: event.summary,
128 failureReason: reason
129 )
130 )
131 continue
132 }
133
134 queue.append(QueuedDelivery(integrationID: target.id, event: event))
135 appendRecord(
136 DeliveryRecord(
137 integrationID: target.id,
138 eventID: event.id,
139 status: .pending,
140 destination: destinationLabel(for: target),
141 summary: event.summary
142 )
143 )
144 }
145 }
146 persistQueue()
147 scheduleProcessing()
148 }
149
150 func recordNoOutboundEvents(for runSummary: String) {
151 let eligibleTargets = targets.filter(\.isEnabled)
152 for target in eligibleTargets {
153 appendRecord(
154 DeliveryRecord(
155 integrationID: target.id,
156 eventID: UUID(),
157 status: .skipped,
158 destination: destinationLabel(for: target),
159 summary: runSummary,
160 failureReason: "Monitoring run produced no outbound events."
161 )
162 )
163 }
164 }
165
166 func sendTest(for targetID: UUID) {
167 guard let target = targets.first(where: { $0.id == targetID }) else { return }
168 let event = MonitoringEvent(
169 type: .test,
170 severity: .info,
171 domain: "example.com",
172 summary: "DomainDig integration test",
173 details: [
174 "source": "manual test",
175 "environment": "local-first"
176 ]
177 )
178
179 // Real events skip a disabled target, so a test event must too —
180 // otherwise a test succeeds against a target that silently drops
181 // everything monitoring sends it.
182 guard target.isEnabled else {
183 appendRecord(
184 DeliveryRecord(
185 integrationID: target.id,
186 eventID: event.id,
187 status: .skipped,
188 destination: destinationLabel(for: target),
189 summary: event.summary,
190 failureReason: Self.disabledTargetReason
191 )
192 )
193 return
194 }
195
196 queue.append(QueuedDelivery(integrationID: target.id, event: event))
197 appendRecord(
198 DeliveryRecord(
199 integrationID: target.id,
200 eventID: event.id,
201 status: .pending,
202 destination: destinationLabel(for: target),
203 summary: event.summary
204 )
205 )
206 persistQueue()
207 scheduleProcessing()
208 }
209
210 /// Restarting the processing task alone leaves any item still in retry
211 /// backoff undue, so the loop would skip it and sleep again. Pulling every
212 /// queued item forward is what makes this button mean "now".
213 func processQueueNow() {
214 guard !queue.isEmpty else {
215 statusMessage = "No deliveries are waiting."
216 return
217 }
218
219 let now = Date()
220 for index in queue.indices {
221 queue[index].nextAttemptAt = now
222 }
223 persistQueue()
224 scheduleProcessing(force: true)
225 }
226
227 func localSecretReferences() -> [String] {
228 targets.compactMap { target in
229 switch target.configuration {
230 case .webhook(let configuration):
231 configuration.credentialReference
232 case .slack(let configuration):
233 configuration.credentialReference
234 case .email(let configuration):
235 configuration.credentialReference
236 }
237 }
238 }
239
240 func resetAfterLocalWipe() {
241 processingTask?.cancel()
242 processingTask = nil
243 targets = []
244 deliveryRecords = []
245 queue = []
246 statusMessage = nil
247 }
248
249 private func scheduleProcessing(force: Bool = false) {
250 if force {
251 processingTask?.cancel()
252 processingTask = nil
253 }
254 guard processingTask == nil else { return }
255 processingTask = Task { [weak self] in
256 guard let self else { return }
257 await self.processQueueLoop()
258 }
259 }
260
261 private func processQueueLoop() async {
262 defer { processingTask = nil }
263
264 while true {
265 let dueItems = queue
266 .enumerated()
267 .filter { $0.element.nextAttemptAt <= Date() }
268
269 if dueItems.isEmpty {
270 guard let nextAttemptAt = queue.map(\.nextAttemptAt).min() else {
271 break
272 }
273
274 let delay = max(0.25, nextAttemptAt.timeIntervalSinceNow)
275 do {
276 try await Task.sleep(nanoseconds: UInt64(delay * 1_000_000_000))
277 continue
278 } catch {
279 break
280 }
281 }
282
283 for entry in dueItems.reversed() {
284 guard entry.offset < queue.count else { continue }
285 let item = queue[entry.offset]
286 await process(item: item, at: entry.offset)
287 }
288 }
289 }
290
291 private func process(item: QueuedDelivery, at index: Int) async {
292 guard let target = targets.first(where: { $0.id == item.integrationID }) else {
293 queue.remove(at: index)
294 persistQueue()
295 return
296 }
297
298 guard item.expiresAt > Date() else {
299 queue.remove(at: index)
300 persistQueue()
301 appendRecord(
302 DeliveryRecord(
303 integrationID: target.id,
304 eventID: item.event.id,
305 status: .expired,
306 destination: destinationLabel(for: target),
307 summary: item.event.summary,
308 failureReason: "Delivery expired before succeeding.",
309 attemptCount: item.attemptCount
310 )
311 )
312 return
313 }
314
315 do {
316 try await deliver(item.event, to: target)
317 queue.remove(at: index)
318 persistQueue()
319 appendRecord(
320 DeliveryRecord(
321 integrationID: target.id,
322 eventID: item.event.id,
323 status: .delivered,
324 destination: destinationLabel(for: target),
325 summary: item.event.summary,
326 attemptCount: item.attemptCount + 1
327 )
328 )
329 statusMessage = "Delivered \(item.event.summary)."
330 } catch {
331 var updated = item
332 updated.attemptCount += 1
333 updated.lastError = error.localizedDescription
334
335 if updated.attemptCount >= 5 {
336 queue.remove(at: index)
337 appendRecord(
338 DeliveryRecord(
339 integrationID: target.id,
340 eventID: item.event.id,
341 status: .failed,
342 destination: destinationLabel(for: target),
343 summary: item.event.summary,
344 failureReason: error.localizedDescription,
345 attemptCount: updated.attemptCount
346 )
347 )
348 } else {
349 let backoff = min(pow(2, Double(updated.attemptCount)) * 30, 3600)
350 updated.nextAttemptAt = Date().addingTimeInterval(backoff)
351 queue[index] = updated
352 appendRecord(
353 DeliveryRecord(
354 integrationID: target.id,
355 eventID: item.event.id,
356 status: .retrying,
357 destination: destinationLabel(for: target),
358 summary: item.event.summary,
359 failureReason: error.localizedDescription,
360 attemptCount: updated.attemptCount
361 )
362 )
363 }
364
365 persistQueue()
366 statusMessage = error.localizedDescription
367 }
368 }
369
370 private func deliver(_ event: MonitoringEvent, to target: IntegrationTarget) async throws {
371 switch target.configuration {
372 case .webhook(let configuration):
373 guard let reference = configuration.credentialReference else {
374 throw IntegrationError.missingSecret
375 }
376 let webhookURLString = try IntegrationSecretStore.secret(reference: reference)
377 try await HTTPIntegrationClient.sendJSON(
378 payload: IntegrationEventPayload(event: event),
379 to: webhookURLString,
380 headers: configuration.additionalHeaders,
381 timeoutSeconds: configuration.timeoutSeconds
382 )
383 case .slack(let configuration):
384 guard let reference = configuration.credentialReference else {
385 throw IntegrationError.missingSecret
386 }
387 let webhookURLString = try IntegrationSecretStore.secret(reference: reference)
388 try await HTTPIntegrationClient.sendJSON(
389 payload: SlackPayload(event: event),
390 to: webhookURLString,
391 headers: [:],
392 timeoutSeconds: 15
393 )
394 case .email(let configuration):
395 guard let reference = configuration.credentialReference else {
396 throw IntegrationError.missingSecret
397 }
398 let password = try IntegrationSecretStore.secret(reference: reference)
399 try await SMTPClient.send(
400 event: event,
401 configuration: configuration,
402 password: password
403 )
404 }
405 }
406
407 private func appendRecord(_ record: DeliveryRecord) {
408 deliveryRecords.insert(record, at: 0)
409 deliveryRecords = Array(deliveryRecords.prefix(250))
410 persistRecords()
411 }
412
413 private func persistTargets() {
414 Self.save(targets, key: StorageKey.targets, defaults: defaults)
415 }
416
417 private func persistRecords() {
418 Self.save(deliveryRecords, key: StorageKey.records, defaults: defaults)
419 }
420
421 private func persistQueue() {
422 Self.save(queue, key: StorageKey.queue, defaults: defaults)
423 }
424
425 private func deleteSecrets(for target: IntegrationTarget) {
426 switch target.configuration {
427 case .webhook(let configuration):
428 if let reference = configuration.credentialReference {
429 try? IntegrationSecretStore.delete(reference: reference)
430 }
431 case .slack(let configuration):
432 if let reference = configuration.credentialReference {
433 try? IntegrationSecretStore.delete(reference: reference)
434 }
435 case .email(let configuration):
436 if let reference = configuration.credentialReference {
437 try? IntegrationSecretStore.delete(reference: reference)
438 }
439 }
440 }
441
442 private func destinationLabel(for target: IntegrationTarget) -> String {
443 switch target.configuration {
444 case .webhook(let configuration):
445 return configuration.endpointDisplayHost.isEmpty ? target.name : configuration.endpointDisplayHost
446 case .slack(let configuration):
447 return configuration.destinationLabel
448 case .email(let configuration):
449 return configuration.recipientAddresses.joined(separator: ", ")
450 }
451 }
452
453 private func filterMismatchReason(for event: MonitoringEvent, target: IntegrationTarget) -> String? {
454 let filters = target.filters
455
456 if event.severity < filters.minimumSeverity {
457 return "Filtered by severity. Event was \(event.severity.title), target requires \(filters.minimumSeverity.title)."
458 }
459
460 if !filters.eventTypes.isEmpty, !filters.eventTypes.contains(event.type) {
461 return "Filtered by event type. Event was \(event.type.title)."
462 }
463
464 if !filters.domains.isEmpty, !filters.domains.map({ $0.lowercased() }).contains(event.domain.lowercased()) {
465 return "Filtered by domain. Event was for \(event.domain)."
466 }
467
468 return nil
469 }
470
471 private static func hostLabel(from string: String) -> String {
472 URL(string: string)?.host ?? "Configured"
473 }
474
475 private static let disabledTargetReason = "This integration is disabled."
476
477 private static func secretReference(for integrationID: UUID, suffix: String) -> String {
478 "integration.\(integrationID.uuidString).\(suffix)"
479 }
480
481 private static func validateHTTPS(_ string: String) throws {
482 guard let url = URL(string: string) else {
483 throw IntegrationError.invalidURL
484 }
485 guard url.scheme?.lowercased() == "https" else {
486 throw IntegrationError.insecureURL
487 }
488 }
489
490 private static func loadTargets(defaults: UserDefaults) -> [IntegrationTarget] {
491 load([IntegrationTarget].self, key: StorageKey.targets, defaults: defaults) ?? []
492 }
493
494 private static func loadRecords(defaults: UserDefaults) -> [DeliveryRecord] {
495 load([DeliveryRecord].self, key: StorageKey.records, defaults: defaults) ?? []
496 }
497
498 private static func loadQueue(defaults: UserDefaults) -> [QueuedDelivery] {
499 load([QueuedDelivery].self, key: StorageKey.queue, defaults: defaults) ?? []
500 }
501
502 private static func load<T: Decodable>(_ type: T.Type, key: String, defaults: UserDefaults) -> T? {
503 guard let data = defaults.data(forKey: key) else {
504 return nil
505 }
506 return try? JSONDecoder().decode(type, from: data)
507 }
508
509 private static func save<T: Encodable>(_ value: T, key: String, defaults: UserDefaults) {
510 if let data = try? JSONEncoder().encode(value) {
511 defaults.set(data, forKey: key)
512 }
513 }
514
515 private enum StorageKey {
516 static let targets = "integrations.targets"
517 static let records = "integrations.records"
518 static let queue = "integrations.queue"
519 }
520}
521
522private struct IntegrationEventPayload: Encodable {
523 let eventType: String
524 let domain: String
525 let timestamp: Date
526 let severity: String
527 let summary: String
528 let details: [String: String]
529
530 init(event: MonitoringEvent) {
531 self.eventType = event.type.rawValue
532 self.domain = event.domain
533 self.timestamp = event.timestamp
534 self.severity = event.severity.rawValue
535 self.summary = event.summary
536 self.details = event.details
537 }
538}
539
540private struct SlackPayload: Encodable {
541 let text: String
542 let blocks: [SlackBlock]
543
544 init(event: MonitoringEvent) {
545 let title = "\(event.severity.title.uppercased()) • \(event.domain)"
546 let detailLines = event.details
547 .sorted { $0.key < $1.key }
548 .prefix(6)
549 .map { "\($0.key): \($0.value)" }
550 .joined(separator: "\n")
551
552 self.text = "\(title) — \(event.summary)"
553 self.blocks = [
554 SlackBlock(
555 type: "section",
556 text: .init(type: "mrkdwn", text: "*\(title)*\n\(event.summary)")
557 ),
558 SlackBlock(
559 type: "section",
560 text: .init(
561 type: "mrkdwn",
562 text: "*Event*: \(event.type.title)\n*Timestamp*: \(event.timestamp.formatted(date: .abbreviated, time: .shortened))"
563 )
564 ),
565 SlackBlock(
566 type: "section",
567 text: .init(type: "mrkdwn", text: detailLines.isEmpty ? "_No extra details_" : detailLines)
568 )
569 ]
570 }
571}
572
573private struct SlackBlock: Encodable {
574 let type: String
575 let text: SlackText
576}
577
578private struct SlackText: Encodable {
579 let type: String
580 let text: String
581}
582
583private enum IntegrationError: LocalizedError {
584 case invalidURL
585 case insecureURL
586 case missingSecret
587 case invalidResponse(Int)
588 case invalidSMTPPort
589 case smtp(String)
590 case streamClosed
591
592 var errorDescription: String? {
593 switch self {
594 case .invalidURL:
595 return "The integration URL is invalid."
596 case .insecureURL:
597 return "The integration URL must use https. A webhook URL is itself a secret, so http would send it in cleartext."
598 case .missingSecret:
599 return "This integration is missing a saved secret."
600 case .invalidResponse(let statusCode):
601 return "The remote endpoint returned \(statusCode)."
602 case .invalidSMTPPort:
603 return "The SMTP port is invalid."
604 case .smtp(let message):
605 return message
606 case .streamClosed:
607 return "The SMTP connection closed unexpectedly."
608 }
609 }
610}
611
612private enum HTTPIntegrationClient {
613 static func sendJSON<T: Encodable>(
614 payload: T,
615 to urlString: String,
616 headers: [String: String],
617 timeoutSeconds: Double
618 ) async throws {
619 guard let url = URL(string: urlString) else {
620 throw IntegrationError.invalidURL
621 }
622 guard url.scheme?.lowercased() == "https" else {
623 throw IntegrationError.insecureURL
624 }
625
626 var request = URLRequest(url: url, timeoutInterval: timeoutSeconds)
627 request.httpMethod = "POST"
628 request.setValue("application/json", forHTTPHeaderField: "Content-Type")
629 for (key, value) in headers {
630 request.setValue(value, forHTTPHeaderField: key)
631 }
632
633 let encoder = JSONEncoder()
634 encoder.dateEncodingStrategy = .iso8601
635 request.httpBody = try encoder.encode(payload)
636
637 let (_, response) = try await URLSession.shared.data(for: request)
638 guard let httpResponse = response as? HTTPURLResponse else {
639 throw IntegrationError.invalidResponse(-1)
640 }
641 guard (200..<300).contains(httpResponse.statusCode) else {
642 throw IntegrationError.invalidResponse(httpResponse.statusCode)
643 }
644 }
645}
646
647private enum IntegrationSecretStore {
648 static func save(secret: String, reference: String) throws {
649 let data = Data(secret.utf8)
650 try? delete(reference: reference)
651
652 let query: [String: Any] = [
653 kSecClass as String: kSecClassGenericPassword,
654 kSecAttrAccount as String: reference,
655 kSecValueData as String: data,
656 kSecAttrAccessible as String: kSecAttrAccessibleAfterFirstUnlock
657 ]
658
659 let status = SecItemAdd(query as CFDictionary, nil)
660 guard status == errSecSuccess else {
661 throw IntegrationError.smtp("Could not save integration secret.")
662 }
663 }
664
665 static func secret(reference: String) throws -> String {
666 let query: [String: Any] = [
667 kSecClass as String: kSecClassGenericPassword,
668 kSecAttrAccount as String: reference,
669 kSecReturnData as String: true,
670 kSecMatchLimit as String: kSecMatchLimitOne
671 ]
672
673 var result: CFTypeRef?
674 let status = SecItemCopyMatching(query as CFDictionary, &result)
675 guard status == errSecSuccess,
676 let data = result as? Data,
677 let secret = String(data: data, encoding: .utf8) else {
678 throw IntegrationError.missingSecret
679 }
680
681 return secret
682 }
683
684 static func delete(reference: String) throws {
685 let query: [String: Any] = [
686 kSecClass as String: kSecClassGenericPassword,
687 kSecAttrAccount as String: reference
688 ]
689 SecItemDelete(query as CFDictionary)
690 }
691}
692
693private enum SMTPClient {
694 static func send(
695 event: MonitoringEvent,
696 configuration: EmailIntegrationConfiguration,
697 password: String
698 ) async throws {
699 guard let port = NWEndpoint.Port(rawValue: UInt16(configuration.port)) else {
700 throw IntegrationError.invalidSMTPPort
701 }
702
703 let parameters: NWParameters = {
704 switch configuration.securityMode {
705 case .plain:
706 return .tcp
707 case .directTLS:
708 let tls = NWProtocolTLS.Options()
709 return NWParameters(tls: tls, tcp: NWProtocolTCP.Options())
710 }
711 }()
712
713 let channel = SMTPChannel(host: configuration.smtpHost, port: port, parameters: parameters)
714 try await channel.start()
715 _ = try await channel.readResponse(expecting: [220])
716 _ = try await channel.sendCommand("EHLO domaindig.local", expecting: [250])
717
718 if !configuration.username.isEmpty {
719 _ = try await channel.sendCommand("AUTH LOGIN", expecting: [334])
720 _ = try await channel.sendCommand(Data(configuration.username.utf8).base64EncodedString(), expecting: [334])
721 _ = try await channel.sendCommand(Data(password.utf8).base64EncodedString(), expecting: [235])
722 }
723
724 _ = try await channel.sendCommand("MAIL FROM:<\(configuration.senderAddress)>", expecting: [250])
725 for recipient in configuration.recipientAddresses {
726 _ = try await channel.sendCommand("RCPT TO:<\(recipient)>", expecting: [250, 251])
727 }
728 _ = try await channel.sendCommand("DATA", expecting: [354])
729
730 let detailLines = event.details
731 .sorted { $0.key < $1.key }
732 .map { "\($0.key): \($0.value)" }
733 .joined(separator: "\r\n")
734 let body = [
735 "From: DomainDig <\(configuration.senderAddress)>",
736 "To: \(configuration.recipientAddresses.joined(separator: ", "))",
737 "Subject: [DomainDig] \(event.severity.title) \(event.domain) \(event.type.title)",
738 "Date: \(DateFormatter.rfc2822.string(from: Date()))",
739 "",
740 event.summary,
741 "",
742 "Domain: \(event.domain)",
743 "Severity: \(event.severity.title)",
744 "Event: \(event.type.title)",
745 "Timestamp: \(event.timestamp.formatted(date: .abbreviated, time: .shortened))",
746 detailLines
747 ]
748 .joined(separator: "\r\n")
749
750 try await channel.sendRaw(body + "\r\n.\r\n")
751 _ = try await channel.readResponse(expecting: [250])
752 _ = try await channel.sendCommand("QUIT", expecting: [221])
753 await channel.cancel()
754 }
755}
756
757/// An actor, not a MainActor class. The previous shape inherited the project's
758/// MainActor default while running its receive loop on a background dispatch
759/// queue, so `parsedLines`/`lineWaiters`/`receiveBuffer` were declared
760/// main-actor-protected and mutated off it — concurrent mutation while resuming
761/// a `CheckedContinuation` can double-resume, which traps. The actor serialises
762/// all of it and forces the Network callbacks to hop in explicitly. (Issue #27.)
763private actor SMTPChannel {
764 private let connection: NWConnection
765 private var parsedLines: [String] = []
766 private var lineWaiters: [CheckedContinuation<String, Error>] = []
767 private var receiveBuffer = Data()
768
769 init(host: String, port: NWEndpoint.Port, parameters: NWParameters) {
770 connection = NWConnection(host: NWEndpoint.Host(host), port: port, using: parameters)
771 }
772
773 func start() async throws {
774 try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
775 // The state handler can fire `.ready` and later `.failed` (or
776 // `.failed` twice); resuming a continuation twice traps. The lock
777 // also keeps the closure Sendable-clean without touching actor state
778 // from the connection's queue.
779 let hasResumed = OSAllocatedUnfairLock(initialState: false)
780 connection.stateUpdateHandler = { state in
781 let isFirst: () -> Bool = {
782 hasResumed.withLock { resumed in
783 if resumed { return false }
784 resumed = true
785 return true
786 }
787 }
788 switch state {
789 case .ready:
790 if isFirst() { continuation.resume() }
791 case .failed(let error):
792 if isFirst() { continuation.resume(throwing: error) }
793 case .cancelled:
794 if isFirst() { continuation.resume(throwing: IntegrationError.streamClosed) }
795 default:
796 break
797 }
798 }
799 connection.start(queue: .global(qos: .utility))
800 }
801 // Started from the actor once the connection is ready, replacing the old
802 // dispatch-queue hop. TCP buffers anything that arrives in the gap.
803 startReceiveLoop()
804 }
805
806 func cancel() {
807 connection.cancel()
808 }
809
810 func sendCommand(_ command: String, expecting codes: Set<Int>) async throws -> String {
811 try await sendRaw(command + "\r\n")
812 return try await readResponse(expecting: codes)
813 }
814
815 func sendCommand(_ command: String, expecting codes: [Int]) async throws -> String {
816 try await sendCommand(command, expecting: Set(codes))
817 }
818
819 func sendRaw(_ string: String) async throws {
820 let data = Data(string.utf8)
821 try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
822 connection.send(content: data, completion: .contentProcessed { error in
823 if let error {
824 continuation.resume(throwing: error)
825 } else {
826 continuation.resume()
827 }
828 })
829 }
830 }
831
832 func readResponse(expecting codes: Set<Int>) async throws -> String {
833 var lines: [String] = []
834
835 while true {
836 let line = try await readLine()
837 lines.append(line)
838
839 guard line.count >= 4,
840 let code = Int(line.prefix(3)) else {
841 continue
842 }
843
844 let delimiterIndex = line.index(line.startIndex, offsetBy: 3)
845 if line[delimiterIndex] == " " {
846 guard codes.contains(code) else {
847 throw IntegrationError.smtp(line)
848 }
849 return lines.joined(separator: "\n")
850 }
851 }
852 }
853
854 private func readLine() async throws -> String {
855 if !parsedLines.isEmpty {
856 return parsedLines.removeFirst()
857 }
858
859 return try await withCheckedThrowingContinuation { continuation in
860 lineWaiters.append(continuation)
861 }
862 }
863
864 private func startReceiveLoop() {
865 // The completion runs on the connection's queue; hop back onto the
866 // actor before touching any state.
867 connection.receive(minimumIncompleteLength: 1, maximumLength: 4096) { [weak self] data, _, isComplete, error in
868 guard let self else { return }
869 Task {
870 await self.handleReceive(data: data, isComplete: isComplete, error: error)
871 }
872 }
873 }
874
875 private func handleReceive(data: Data?, isComplete: Bool, error: Error?) {
876 if let error {
877 failWaiters(with: error)
878 return
879 }
880
881 if let data, !data.isEmpty {
882 receiveBuffer.append(data)
883 flushBuffer()
884 }
885
886 if isComplete {
887 failWaiters(with: IntegrationError.streamClosed)
888 return
889 }
890
891 startReceiveLoop()
892 }
893
894 private func flushBuffer() {
895 let delimiter = Data("\r\n".utf8)
896 while let range = receiveBuffer.range(of: delimiter) {
897 let lineData = receiveBuffer.subdata(in: receiveBuffer.startIndex..<range.lowerBound)
898 receiveBuffer.removeSubrange(receiveBuffer.startIndex..<range.upperBound)
899 let line = String(data: lineData, encoding: .utf8) ?? ""
900 if !lineWaiters.isEmpty {
901 let continuation = lineWaiters.removeFirst()
902 continuation.resume(returning: line)
903 } else {
904 parsedLines.append(line)
905 }
906 }
907 }
908
909 private func failWaiters(with error: Error) {
910 let waiters = lineWaiters
911 lineWaiters.removeAll()
912 for waiter in waiters {
913 waiter.resume(throwing: error)
914 }
915 }
916}
917
918private extension DateFormatter {
919 static let rfc2822: DateFormatter = {
920 let formatter = DateFormatter()
921 formatter.locale = Locale(identifier: "en_US_POSIX")
922 formatter.timeZone = TimeZone(secondsFromGMT: 0)
923 formatter.dateFormat = "EEE, dd MMM yyyy HH:mm:ss Z"
924 return formatter
925 }()
926}