Thomas GüttlerandClaude Sonnet 4.6
16607f7ea0
- New test/integration/email_repository_jmap_test.dart covers full sync, incremental sync, server-side deletion reconciliation, getEmailBody with cache, sendEmail, and flushPendingChanges (flag, delete, move). - Fix: Email/set destroy must be an array per RFC 8620 §5.3; was passing a bare string which Stalwart silently rejected. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1783 lines
56 KiB
Dart
1783 lines
56 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
import 'dart:io';
|
|
import 'dart:math' as math;
|
|
|
|
import 'package:drift/drift.dart';
|
|
import 'package:enough_mail/enough_mail.dart' as imap;
|
|
import 'package:http/http.dart' as http;
|
|
import 'package:path/path.dart' as p;
|
|
import 'package:path_provider/path_provider.dart';
|
|
|
|
import '../../core/models/account.dart' as account_model;
|
|
import '../../core/models/email.dart' as model;
|
|
import '../../core/repositories/account_repository.dart';
|
|
import '../../core/repositories/email_repository.dart';
|
|
import '../../core/utils/logger.dart';
|
|
import '../db/database.dart';
|
|
import '../imap/imap_client_factory.dart';
|
|
import '../jmap/jmap_client.dart';
|
|
|
|
typedef SmtpConnectFn = Future<imap.SmtpClient> Function(
|
|
account_model.Account account,
|
|
String username,
|
|
String password,
|
|
);
|
|
typedef GetCacheDirFn = Future<Directory> Function();
|
|
|
|
class EmailRepositoryImpl implements EmailRepository {
|
|
EmailRepositoryImpl(
|
|
this._db,
|
|
this._accounts, {
|
|
ImapConnectFn imapConnect = connectImap,
|
|
SmtpConnectFn smtpConnect = connectSmtp,
|
|
GetCacheDirFn getCacheDir = getTemporaryDirectory,
|
|
http.Client? httpClient,
|
|
}) : _imapConnect = imapConnect,
|
|
_smtpConnect = smtpConnect,
|
|
_getCacheDir = getCacheDir,
|
|
_httpClient = httpClient ?? http.Client();
|
|
|
|
final AppDatabase _db;
|
|
final AccountRepository _accounts;
|
|
final ImapConnectFn _imapConnect;
|
|
final SmtpConnectFn _smtpConnect;
|
|
final GetCacheDirFn _getCacheDir;
|
|
final http.Client _httpClient;
|
|
|
|
String _effectiveUsername(account_model.Account account) =>
|
|
account.username.isNotEmpty ? account.username : account.email;
|
|
|
|
// ── Observe ────────────────────────────────────────────────────────────────
|
|
|
|
@override
|
|
Stream<List<model.Email>> observeEmails(
|
|
String accountId,
|
|
String mailboxPath,
|
|
) {
|
|
return (_db.select(_db.emails)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(accountId) &
|
|
t.mailboxPath.equals(mailboxPath),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.desc(t.receivedAt)]))
|
|
.watch()
|
|
.map((rows) => rows.map(_toModel).toList());
|
|
}
|
|
|
|
@override
|
|
Future<model.Email?> getEmail(String emailId) async {
|
|
final row = await (_db.select(_db.emails)
|
|
..where((t) => t.id.equals(emailId)))
|
|
.getSingleOrNull();
|
|
return row == null ? null : _toModel(row);
|
|
}
|
|
|
|
// ── Body (on-demand) ───────────────────────────────────────────────────────
|
|
|
|
static const _bodyCacheTtl = Duration(days: 7);
|
|
|
|
@override
|
|
Future<model.EmailBody> getEmailBody(String emailId) async {
|
|
final cached = await (_db.select(_db.emailBodies)
|
|
..where((t) => t.emailId.equals(emailId)))
|
|
.getSingleOrNull();
|
|
if (cached != null) {
|
|
// Re-fetch if cachedAt is null (legacy row) or older than the TTL.
|
|
final age = cached.cachedAt == null
|
|
? _bodyCacheTtl + const Duration(seconds: 1)
|
|
: DateTime.now().difference(cached.cachedAt!);
|
|
if (age <= _bodyCacheTtl) return _bodyRowToModel(cached);
|
|
}
|
|
|
|
final emailRow = await (_db.select(_db.emails)
|
|
..where((t) => t.id.equals(emailId)))
|
|
.getSingle();
|
|
final account = (await _accounts.getAccount(emailRow.accountId))!;
|
|
final password = await _accounts.getPassword(account.id);
|
|
|
|
if (account.type == account_model.AccountType.jmap) {
|
|
return _getEmailBodyJmap(emailId, account, password);
|
|
}
|
|
|
|
final client =
|
|
await _imapConnect(account, _effectiveUsername(account), password);
|
|
try {
|
|
await client.selectMailboxByPath(emailRow.mailboxPath);
|
|
final fetch = await client.uidFetchMessage(emailRow.uid, '(BODY[])');
|
|
final msg = fetch.messages.first;
|
|
final textBody = msg.decodeTextPlainPart();
|
|
final htmlBody = msg.decodeTextHtmlPart();
|
|
final contentInfos = msg.findContentInfo();
|
|
|
|
final attachmentsJson = jsonEncode(
|
|
contentInfos
|
|
.map(
|
|
(a) => {
|
|
'filename': a.fileName ?? '',
|
|
'contentType': a.contentType?.mediaType.text ?? '',
|
|
'size': a.size ?? 0,
|
|
'fetchPartId': a.fetchId,
|
|
},
|
|
)
|
|
.toList(),
|
|
);
|
|
|
|
await _db.into(_db.emailBodies).insertOnConflictUpdate(
|
|
EmailBodiesCompanion.insert(
|
|
emailId: emailId,
|
|
textBody: Value(textBody),
|
|
htmlBody: Value(htmlBody),
|
|
attachmentsJson: Value(attachmentsJson),
|
|
cachedAt: Value(DateTime.now()),
|
|
),
|
|
);
|
|
return model.EmailBody(
|
|
emailId: emailId,
|
|
textBody: textBody,
|
|
htmlBody: htmlBody,
|
|
attachments: _parseAttachments(attachmentsJson),
|
|
);
|
|
} finally {
|
|
await client.logout();
|
|
}
|
|
}
|
|
|
|
Future<model.EmailBody> _getEmailBodyJmap(
|
|
String emailId,
|
|
account_model.Account account,
|
|
String password,
|
|
) async {
|
|
final jmapUrl = account.jmapUrl!;
|
|
final jmap = await JmapClient.connect(
|
|
httpClient: _httpClient,
|
|
jmapUrl: Uri.parse(jmapUrl),
|
|
username: _effectiveUsername(account),
|
|
password: password,
|
|
);
|
|
|
|
final jmapEmailId = emailId.contains(':')
|
|
? emailId.substring(emailId.indexOf(':') + 1)
|
|
: emailId;
|
|
|
|
final responses = await jmap.call([
|
|
[
|
|
'Email/get',
|
|
{
|
|
'accountId': jmap.accountId,
|
|
'ids': [jmapEmailId],
|
|
'properties': [
|
|
'id',
|
|
'textBody',
|
|
'htmlBody',
|
|
'bodyValues',
|
|
'attachments',
|
|
],
|
|
'fetchHTMLBodyValues': true,
|
|
'fetchTextBodyValues': true,
|
|
},
|
|
'0',
|
|
],
|
|
]);
|
|
|
|
final result = _responseArgs(responses, 0, 'Email/get');
|
|
final emailData =
|
|
(result['list'] as List<dynamic>).first as Map<String, dynamic>;
|
|
|
|
final (textBody, htmlBody, attachmentsJson) = _parseJmapBody(emailData);
|
|
|
|
await _db.into(_db.emailBodies).insertOnConflictUpdate(
|
|
EmailBodiesCompanion.insert(
|
|
emailId: emailId,
|
|
textBody: Value(textBody),
|
|
htmlBody: Value(htmlBody),
|
|
attachmentsJson: Value(attachmentsJson),
|
|
cachedAt: Value(DateTime.now()),
|
|
),
|
|
);
|
|
|
|
return model.EmailBody(
|
|
emailId: emailId,
|
|
textBody: textBody,
|
|
htmlBody: htmlBody,
|
|
attachments: _parseAttachments(attachmentsJson),
|
|
);
|
|
}
|
|
|
|
// ── Sync ───────────────────────────────────────────────────────────────────
|
|
|
|
@override
|
|
Future<model.SyncEmailsResult> syncEmails(
|
|
String accountId,
|
|
String mailboxPath,
|
|
) async {
|
|
final account = (await _accounts.getAccount(accountId))!;
|
|
final password = await _accounts.getPassword(accountId);
|
|
switch (account.type) {
|
|
case account_model.AccountType.imap:
|
|
return _syncEmailsImap(account, password, mailboxPath);
|
|
case account_model.AccountType.jmap:
|
|
return _syncEmailsJmap(account, password, mailboxPath);
|
|
}
|
|
}
|
|
|
|
Future<model.SyncEmailsResult> _syncEmailsImap(
|
|
account_model.Account account,
|
|
String password,
|
|
String mailboxPath,
|
|
) async {
|
|
final client =
|
|
await _imapConnect(account, _effectiveUsername(account), password);
|
|
try {
|
|
// Only request CONDSTORE if the server advertises it. Servers that don't
|
|
// support the extension may reject SELECT with (CONDSTORE) with BAD.
|
|
final supportsCondStore = client.serverInfo.supports('CONDSTORE') ||
|
|
client.serverInfo.supports('QRESYNC');
|
|
final selectedMailbox = await client.selectMailboxByPath(
|
|
mailboxPath,
|
|
enableCondStore: supportsCondStore,
|
|
);
|
|
final uidValidity = selectedMailbox.uidValidity ?? 0;
|
|
final serverModSeq = selectedMailbox.highestModSequence;
|
|
final resourceType = 'IMAP:$mailboxPath';
|
|
final checkpoint = await _loadImapCheckpoint(account.id, resourceType);
|
|
|
|
if (checkpoint == null || checkpoint['uidValidity'] != uidValidity) {
|
|
// First run or UID validity changed — full sync.
|
|
if (checkpoint != null) {
|
|
// UID validity changed: remove stale local emails for this mailbox.
|
|
await (_db.delete(_db.emails)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(account.id) &
|
|
t.mailboxPath.equals(mailboxPath),
|
|
))
|
|
.go();
|
|
}
|
|
// Use UID SEARCH ALL + UID FETCH so every message gets a reliable UID.
|
|
// Regular FETCH 1:* may not populate msg.uid on all servers.
|
|
final allUids = (await client.uidSearchMessages(searchCriteria: 'ALL'))
|
|
.matchingSequence
|
|
?.toList() ??
|
|
[];
|
|
var bytes = 0;
|
|
if (allUids.isNotEmpty) {
|
|
bytes = await _fetchAndUpsertImap(
|
|
client,
|
|
account,
|
|
mailboxPath,
|
|
imap.MessageSequence.fromIds(allUids, isUid: true),
|
|
);
|
|
}
|
|
final maxUid = allUids.isEmpty ? 0 : allUids.reduce(math.max);
|
|
await _saveImapCheckpoint(
|
|
account.id,
|
|
resourceType,
|
|
uidValidity,
|
|
maxUid,
|
|
highestModSeq: serverModSeq,
|
|
);
|
|
return model.SyncEmailsResult(
|
|
fetched: allUids.length,
|
|
skipped: 0,
|
|
bytesTransferred: bytes,
|
|
);
|
|
} else {
|
|
// Incremental sync.
|
|
final lastUid = checkpoint['lastUid'] as int;
|
|
final storedModSeq = checkpoint['highestModSeq'] as int?;
|
|
final totalOnServer = selectedMailbox.messagesExists;
|
|
|
|
// CONDSTORE fast-path: nothing has changed on the server.
|
|
if (serverModSeq != null &&
|
|
storedModSeq != null &&
|
|
serverModSeq == storedModSeq) {
|
|
return model.SyncEmailsResult(
|
|
fetched: 0,
|
|
skipped: totalOnServer,
|
|
bytesTransferred: 0,
|
|
);
|
|
}
|
|
|
|
// Fetch new messages.
|
|
final newUids = (await client.uidSearchMessages(
|
|
searchCriteria: 'UID ${lastUid + 1}:*',
|
|
))
|
|
.matchingSequence
|
|
?.toList() ??
|
|
[];
|
|
var bytes = 0;
|
|
if (newUids.isNotEmpty) {
|
|
bytes = await _fetchAndUpsertImap(
|
|
client,
|
|
account,
|
|
mailboxPath,
|
|
imap.MessageSequence.fromIds(newUids, isUid: true),
|
|
);
|
|
}
|
|
|
|
// CONDSTORE flag update: refresh flags only for messages that changed.
|
|
if (serverModSeq != null && storedModSeq != null) {
|
|
await _refreshFlagsImap(client, account, mailboxPath, storedModSeq);
|
|
}
|
|
|
|
// Detect remote deletions.
|
|
final serverUids =
|
|
(await client.uidSearchMessages(searchCriteria: 'ALL'))
|
|
.matchingSequence
|
|
?.toList() ??
|
|
[];
|
|
await _reconcileDeletedImap(account.id, mailboxPath, serverUids);
|
|
final maxUid =
|
|
serverUids.isEmpty ? lastUid : serverUids.reduce(math.max);
|
|
await _saveImapCheckpoint(
|
|
account.id,
|
|
resourceType,
|
|
uidValidity,
|
|
maxUid,
|
|
highestModSeq: serverModSeq,
|
|
);
|
|
return model.SyncEmailsResult(
|
|
fetched: newUids.length,
|
|
skipped: serverUids.length - newUids.length,
|
|
bytesTransferred: bytes,
|
|
);
|
|
}
|
|
} finally {
|
|
await client.logout();
|
|
}
|
|
}
|
|
|
|
/// Fetches FLAGS for all messages modified since [sinceModSeq] and updates
|
|
/// the local DB. Only messages whose modseq is > [sinceModSeq] are returned
|
|
/// by the server (RFC 7162 §3.2).
|
|
Future<void> _refreshFlagsImap(
|
|
imap.ImapClient client,
|
|
account_model.Account account,
|
|
String mailboxPath,
|
|
int sinceModSeq,
|
|
) async {
|
|
final result = await client.uidFetchMessages(
|
|
imap.MessageSequence.fromAll(),
|
|
'FLAGS',
|
|
changedSinceModSequence: sinceModSeq,
|
|
);
|
|
for (final msg in result.messages) {
|
|
final uid = msg.uid;
|
|
if (uid == null) continue;
|
|
final emailId = '${account.id}:$uid';
|
|
await (_db.update(_db.emails)..where((t) => t.id.equals(emailId))).write(
|
|
EmailsCompanion(
|
|
isSeen: Value(msg.flags?.contains(r'\Seen') ?? false),
|
|
isFlagged: Value(msg.flags?.contains(r'\Flagged') ?? false),
|
|
),
|
|
);
|
|
}
|
|
}
|
|
|
|
// Returns the total bytes transferred (sum of RFC822.SIZE for each message).
|
|
Future<int> _fetchAndUpsertImap(
|
|
imap.ImapClient client,
|
|
account_model.Account account,
|
|
String mailboxPath,
|
|
imap.MessageSequence sequence,
|
|
) async {
|
|
final fetch = sequence.isUidSequence
|
|
? await client.uidFetchMessages(
|
|
sequence,
|
|
'(UID FLAGS ENVELOPE BODYSTRUCTURE RFC822.SIZE)',
|
|
)
|
|
: await client.fetchMessages(
|
|
sequence,
|
|
'(UID FLAGS ENVELOPE BODYSTRUCTURE RFC822.SIZE)',
|
|
);
|
|
var bytes = 0;
|
|
await _db.transaction(() async {
|
|
for (final msg in fetch.messages) {
|
|
final envelope = msg.envelope;
|
|
if (envelope == null) {
|
|
log('IMAP: skipping message with no envelope (uid=${msg.uid}, mailbox=$mailboxPath)');
|
|
continue;
|
|
}
|
|
final uid = msg.uid;
|
|
if (uid == null) {
|
|
log('IMAP: skipping message with no uid (mailbox=$mailboxPath)');
|
|
continue;
|
|
}
|
|
bytes += msg.size ?? 0;
|
|
final emailId = '${account.id}:$uid';
|
|
await _db.into(_db.emails).insertOnConflictUpdate(
|
|
EmailsCompanion.insert(
|
|
id: emailId,
|
|
accountId: account.id,
|
|
mailboxPath: mailboxPath,
|
|
uid: uid,
|
|
subject: Value(envelope.subject),
|
|
sentAt: Value(envelope.date),
|
|
receivedAt: envelope.date ?? DateTime.now(),
|
|
fromJson: Value(_encodeAddresses(envelope.from)),
|
|
toAddresses: Value(_encodeAddresses(envelope.to)),
|
|
ccJson: Value(_encodeAddresses(envelope.cc)),
|
|
isSeen: Value(msg.flags?.contains(r'\Seen') ?? false),
|
|
isFlagged: Value(msg.flags?.contains(r'\Flagged') ?? false),
|
|
hasAttachment: Value(msg.hasAttachments()),
|
|
),
|
|
);
|
|
}
|
|
});
|
|
return bytes;
|
|
}
|
|
|
|
Future<Map<String, dynamic>?> _loadImapCheckpoint(
|
|
String accountId,
|
|
String resourceType,
|
|
) async {
|
|
final raw = await _loadSyncState(accountId, resourceType);
|
|
if (raw == null) return null;
|
|
return jsonDecode(raw) as Map<String, dynamic>;
|
|
}
|
|
|
|
Future<void> _saveImapCheckpoint(
|
|
String accountId,
|
|
String resourceType,
|
|
int uidValidity,
|
|
int lastUid, {
|
|
int? highestModSeq,
|
|
}) async {
|
|
final data = <String, dynamic>{
|
|
'uidValidity': uidValidity,
|
|
'lastUid': lastUid,
|
|
};
|
|
if (highestModSeq != null) data['highestModSeq'] = highestModSeq;
|
|
await _saveSyncState(accountId, resourceType, jsonEncode(data));
|
|
}
|
|
|
|
Future<void> _reconcileDeletedImap(
|
|
String accountId,
|
|
String mailboxPath,
|
|
List<int> serverUids,
|
|
) async {
|
|
final serverUidSet = serverUids.toSet();
|
|
final localRows = await (_db.select(_db.emails)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(accountId) &
|
|
t.mailboxPath.equals(mailboxPath),
|
|
))
|
|
.get();
|
|
for (final row in localRows) {
|
|
if (!serverUidSet.contains(row.uid)) {
|
|
await (_db.delete(_db.emails)..where((t) => t.id.equals(row.id))).go();
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── JMAP email sync ────────────────────────────────────────────────────────
|
|
|
|
static const _jmapPageSize = 500;
|
|
|
|
/// Pending changes exceeding this attempt count are evicted rather than
|
|
/// retried, preventing unbounded queue growth from permanent server errors.
|
|
static const _maxChangeAttempts = 5;
|
|
|
|
static const _emailProperties = [
|
|
'id',
|
|
'mailboxIds',
|
|
'subject',
|
|
'sentAt',
|
|
'receivedAt',
|
|
'from',
|
|
'to',
|
|
'cc',
|
|
'keywords',
|
|
'hasAttachment',
|
|
'preview',
|
|
'textBody',
|
|
'htmlBody',
|
|
'bodyValues',
|
|
'attachments',
|
|
];
|
|
|
|
static const _emailGetBodyOptions = {
|
|
'fetchHTMLBodyValues': true,
|
|
'fetchTextBodyValues': true,
|
|
};
|
|
|
|
Future<model.SyncEmailsResult> _syncEmailsJmap(
|
|
account_model.Account account,
|
|
String password,
|
|
String mailboxJmapId,
|
|
) async {
|
|
final jmapUrl = account.jmapUrl;
|
|
if (jmapUrl == null || jmapUrl.isEmpty) {
|
|
throw Exception('JMAP account ${account.id} has no jmapUrl');
|
|
}
|
|
|
|
final jmap = await JmapClient.connect(
|
|
httpClient: _httpClient,
|
|
jmapUrl: Uri.parse(jmapUrl),
|
|
username: _effectiveUsername(account),
|
|
password: password,
|
|
);
|
|
|
|
final storedState = await _loadSyncState(account.id, 'Email');
|
|
|
|
if (storedState == null) {
|
|
return _jmapFullEmailSync(account.id, jmap, mailboxJmapId);
|
|
} else {
|
|
return _jmapIncrementalEmailSync(account.id, jmap, storedState);
|
|
}
|
|
}
|
|
|
|
Future<model.SyncEmailsResult> _jmapFullEmailSync(
|
|
String accountId,
|
|
JmapClient jmap,
|
|
String mailboxJmapId,
|
|
) async {
|
|
int position = 0;
|
|
String? firstState;
|
|
var fetched = 0;
|
|
var bytes = 0;
|
|
|
|
while (true) {
|
|
final responses = await jmap.call([
|
|
[
|
|
'Email/query',
|
|
{
|
|
'accountId': jmap.accountId,
|
|
'filter': {'inMailbox': mailboxJmapId},
|
|
'sort': [
|
|
{'property': 'receivedAt', 'isAscending': false},
|
|
],
|
|
'limit': _jmapPageSize,
|
|
'position': position,
|
|
'calculateTotal': true,
|
|
},
|
|
'0',
|
|
],
|
|
[
|
|
'Email/get',
|
|
{
|
|
'accountId': jmap.accountId,
|
|
'#ids': {'resultOf': '0', 'name': 'Email/query', 'path': '/ids'},
|
|
'properties': _emailProperties,
|
|
..._emailGetBodyOptions,
|
|
},
|
|
'1',
|
|
],
|
|
]);
|
|
|
|
final queryResult = _responseArgs(responses, 0, 'Email/query');
|
|
final ids = queryResult['ids'] as List<dynamic>;
|
|
final total = queryResult['total'] as int?;
|
|
|
|
final getResult = _responseArgs(responses, 1, 'Email/get');
|
|
firstState ??= getResult['state'] as String;
|
|
final list = getResult['list'] as List<dynamic>;
|
|
bytes += await _upsertJmapEmails(accountId, list);
|
|
fetched += list.length;
|
|
|
|
position += ids.length;
|
|
if (ids.isEmpty || total == null || position >= total) break;
|
|
}
|
|
|
|
await _saveSyncState(accountId, 'Email', firstState);
|
|
return model.SyncEmailsResult(
|
|
fetched: fetched,
|
|
skipped: 0,
|
|
bytesTransferred: bytes,
|
|
);
|
|
}
|
|
|
|
Future<model.SyncEmailsResult> _jmapIncrementalEmailSync(
|
|
String accountId,
|
|
JmapClient jmap,
|
|
String sinceState,
|
|
) async {
|
|
final responses = await jmap.call([
|
|
[
|
|
'Email/changes',
|
|
{'accountId': jmap.accountId, 'sinceState': sinceState},
|
|
'0',
|
|
]
|
|
]);
|
|
|
|
final changes = _responseArgs(responses, 0, 'Email/changes');
|
|
final newState = changes['newState'] as String;
|
|
final created = List<String>.from(changes['created'] as List? ?? []);
|
|
final updated = List<String>.from(changes['updated'] as List? ?? []);
|
|
final destroyed = List<String>.from(changes['destroyed'] as List? ?? []);
|
|
|
|
var fetched = 0;
|
|
var bytes = 0;
|
|
final toFetch = [...created, ...updated];
|
|
if (toFetch.isNotEmpty) {
|
|
final getResponses = await jmap.call([
|
|
[
|
|
'Email/get',
|
|
{
|
|
'accountId': jmap.accountId,
|
|
'ids': toFetch,
|
|
'properties': _emailProperties,
|
|
..._emailGetBodyOptions,
|
|
},
|
|
'1',
|
|
]
|
|
]);
|
|
final getResult = _responseArgs(getResponses, 0, 'Email/get');
|
|
final list = getResult['list'] as List<dynamic>;
|
|
bytes += await _upsertJmapEmails(accountId, list);
|
|
fetched += list.length;
|
|
}
|
|
|
|
for (final jmapId in destroyed) {
|
|
await (_db.delete(_db.emails)
|
|
..where((t) => t.id.equals('$accountId:$jmapId')))
|
|
.go();
|
|
}
|
|
|
|
await _saveSyncState(accountId, 'Email', newState);
|
|
return model.SyncEmailsResult(
|
|
fetched: fetched,
|
|
skipped: 0,
|
|
bytesTransferred: bytes,
|
|
);
|
|
}
|
|
|
|
// Returns total bytes transferred (sum of JMAP `size` fields).
|
|
Future<int> _upsertJmapEmails(String accountId, List<dynamic> emails) async {
|
|
var bytes = 0;
|
|
for (final e in emails) {
|
|
final m = e as Map<String, dynamic>;
|
|
final jmapId = m['id'] as String;
|
|
final dbId = '$accountId:$jmapId';
|
|
bytes += (m['size'] as int?) ?? 0;
|
|
|
|
// Use first mailbox ID as the primary mailboxPath.
|
|
final mailboxIds = m['mailboxIds'] as Map<String, dynamic>?;
|
|
final mailboxPath = mailboxIds?.keys.firstOrNull ?? '';
|
|
|
|
final keywords = m['keywords'] as Map<String, dynamic>? ?? {};
|
|
final from = _encodeJmapAddresses(m['from']);
|
|
final to = _encodeJmapAddresses(m['to']);
|
|
final cc = _encodeJmapAddresses(m['cc']);
|
|
final sentAt = _parseDate(m['sentAt'] as String?);
|
|
final receivedAt =
|
|
_parseDate(m['receivedAt'] as String?) ?? DateTime.now();
|
|
|
|
await _db.into(_db.emails).insertOnConflictUpdate(
|
|
EmailsCompanion.insert(
|
|
id: dbId,
|
|
accountId: accountId,
|
|
mailboxPath: mailboxPath,
|
|
uid: 0, // not used for JMAP accounts
|
|
subject: Value(m['subject'] as String?),
|
|
sentAt: Value(sentAt),
|
|
receivedAt: receivedAt,
|
|
fromJson: Value(from),
|
|
toAddresses: Value(to),
|
|
ccJson: Value(cc),
|
|
preview: Value(m['preview'] as String?),
|
|
isSeen: Value(keywords.containsKey(r'$seen')),
|
|
isFlagged: Value(keywords.containsKey(r'$flagged')),
|
|
hasAttachment: Value((m['hasAttachment'] as bool?) ?? false),
|
|
),
|
|
);
|
|
|
|
// Cache body if the server included bodyValues in this response.
|
|
if (m.containsKey('bodyValues')) {
|
|
final (textBody, htmlBody, attachmentsJson) = _parseJmapBody(m);
|
|
await _db.into(_db.emailBodies).insertOnConflictUpdate(
|
|
EmailBodiesCompanion.insert(
|
|
emailId: dbId,
|
|
textBody: Value(textBody),
|
|
htmlBody: Value(htmlBody),
|
|
attachmentsJson: Value(attachmentsJson),
|
|
cachedAt: Value(DateTime.now()),
|
|
),
|
|
);
|
|
}
|
|
}
|
|
return bytes;
|
|
}
|
|
|
|
/// Extracts text body, HTML body, and attachments JSON from a JMAP Email object
|
|
/// that was fetched with fetchHTMLBodyValues/fetchTextBodyValues.
|
|
(String? textBody, String? htmlBody, String attachmentsJson) _parseJmapBody(
|
|
Map<String, dynamic> m,
|
|
) {
|
|
final bodyValues = m['bodyValues'] as Map<String, dynamic>? ?? {};
|
|
final textBodyParts = m['textBody'] as List<dynamic>? ?? [];
|
|
final htmlBodyParts = m['htmlBody'] as List<dynamic>? ?? [];
|
|
final jmapAttachments = m['attachments'] as List<dynamic>? ?? [];
|
|
|
|
String? textBody;
|
|
if (textBodyParts.isNotEmpty) {
|
|
final partId =
|
|
(textBodyParts.first as Map<String, dynamic>)['partId'] as String?;
|
|
if (partId != null) {
|
|
textBody =
|
|
(bodyValues[partId] as Map<String, dynamic>?)?['value'] as String?;
|
|
}
|
|
}
|
|
|
|
String? htmlBody;
|
|
if (htmlBodyParts.isNotEmpty) {
|
|
final partId =
|
|
(htmlBodyParts.first as Map<String, dynamic>)['partId'] as String?;
|
|
if (partId != null) {
|
|
htmlBody =
|
|
(bodyValues[partId] as Map<String, dynamic>?)?['value'] as String?;
|
|
}
|
|
}
|
|
|
|
final attachmentsJson = jsonEncode(
|
|
jmapAttachments.map((a) {
|
|
final att = a as Map<String, dynamic>;
|
|
return {
|
|
'filename': att['name'] ?? '',
|
|
'contentType': att['type'] ?? '',
|
|
'size': att['size'] ?? 0,
|
|
'fetchPartId': att['blobId'] ?? '',
|
|
};
|
|
}).toList(),
|
|
);
|
|
|
|
return (textBody, htmlBody, attachmentsJson);
|
|
}
|
|
|
|
// ── Pending-change helpers ────────────────────────────────────────────────
|
|
|
|
/// Records a failure for [row]: increments attempt count and stores the
|
|
/// error message. When attempts reach [_maxChangeAttempts] the row is
|
|
/// deleted instead — the change is permanently abandoned.
|
|
Future<void> _recordChangeError(PendingChangeRow row, Object error) async {
|
|
final next = row.attempts + 1;
|
|
if (next >= _maxChangeAttempts) {
|
|
await (_db.delete(_db.pendingChanges)..where((t) => t.id.equals(row.id)))
|
|
.go();
|
|
} else {
|
|
await (_db.update(_db.pendingChanges)..where((t) => t.id.equals(row.id)))
|
|
.write(
|
|
PendingChangesCompanion(
|
|
attempts: Value(next),
|
|
lastError: Value(error.toString()),
|
|
),
|
|
);
|
|
}
|
|
}
|
|
|
|
// ── sync_state helpers ────────────────────────────────────────────────────
|
|
|
|
Future<String?> _loadSyncState(String accountId, String resourceType) async {
|
|
final row = await (_db.select(_db.syncStates)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(accountId) &
|
|
t.resourceType.equals(resourceType),
|
|
))
|
|
.getSingleOrNull();
|
|
return row?.state;
|
|
}
|
|
|
|
Future<void> _saveSyncState(
|
|
String accountId,
|
|
String resourceType,
|
|
String state,
|
|
) async {
|
|
await _db.into(_db.syncStates).insertOnConflictUpdate(
|
|
SyncStatesCompanion.insert(
|
|
accountId: accountId,
|
|
resourceType: resourceType,
|
|
state: state,
|
|
syncedAt: DateTime.now(),
|
|
),
|
|
);
|
|
}
|
|
|
|
// ── JMAP push ────────────────────────────────────────────────────────────
|
|
|
|
@override
|
|
Stream<void> watchJmapPush(String accountId, String password) {
|
|
final controller = StreamController<void>();
|
|
StreamSubscription<String>? innerSub;
|
|
|
|
controller.onCancel = () => innerSub?.cancel();
|
|
|
|
unawaited(() async {
|
|
try {
|
|
final account = await _accounts.getAccount(accountId);
|
|
if (account == null || account.type != account_model.AccountType.jmap) {
|
|
await controller.close();
|
|
return;
|
|
}
|
|
|
|
final jmapUrl = account.jmapUrl;
|
|
if (jmapUrl == null || jmapUrl.isEmpty) {
|
|
await controller.close();
|
|
return;
|
|
}
|
|
|
|
final JmapClient jmap;
|
|
try {
|
|
jmap = await JmapClient.connect(
|
|
httpClient: _httpClient,
|
|
jmapUrl: Uri.parse(jmapUrl),
|
|
username: _effectiveUsername(account),
|
|
password: password,
|
|
);
|
|
} catch (_) {
|
|
await controller.close();
|
|
return;
|
|
}
|
|
|
|
final sseUrl = jmap.eventSourceUrl;
|
|
if (sseUrl == null) {
|
|
await controller.close();
|
|
return;
|
|
}
|
|
|
|
final credentials = base64
|
|
.encode(utf8.encode('${_effectiveUsername(account)}:$password'));
|
|
|
|
http.StreamedResponse response;
|
|
try {
|
|
final request = http.Request('GET', Uri.parse(sseUrl));
|
|
request.headers['Accept'] = 'text/event-stream';
|
|
request.headers['Authorization'] = 'Basic $credentials';
|
|
response = await _httpClient
|
|
.send(request)
|
|
.timeout(const Duration(seconds: 10));
|
|
if (response.statusCode != 200) {
|
|
await controller.close();
|
|
return;
|
|
}
|
|
} catch (_) {
|
|
await controller.close();
|
|
return;
|
|
}
|
|
|
|
var buffer = '';
|
|
innerSub = response.stream
|
|
.transform(utf8.decoder)
|
|
.timeout(const Duration(minutes: 25))
|
|
.listen(
|
|
(chunk) {
|
|
buffer += chunk;
|
|
final lines = buffer.split('\n');
|
|
buffer = lines.removeLast();
|
|
for (final line in lines) {
|
|
if (!line.startsWith('data:')) continue;
|
|
final data = line.substring(5).trim();
|
|
try {
|
|
final decoded = jsonDecode(data) as Map<String, dynamic>;
|
|
if (decoded['@type'] == 'StateChange') {
|
|
controller.add(null);
|
|
}
|
|
} catch (_) {
|
|
// Malformed JSON — ignore line
|
|
}
|
|
}
|
|
},
|
|
onDone: () => controller.close(),
|
|
onError: (_) => controller.close(),
|
|
cancelOnError: true,
|
|
);
|
|
} catch (_) {
|
|
await controller.close();
|
|
}
|
|
}());
|
|
|
|
return controller.stream;
|
|
}
|
|
|
|
// ── JMAP helpers ─────────────────────────────────────────────────────────
|
|
|
|
Map<String, dynamic> _responseArgs(
|
|
List<dynamic> responses,
|
|
int index,
|
|
String expectedMethod,
|
|
) {
|
|
final triple = responses[index] as List<dynamic>;
|
|
final method = triple[0] as String;
|
|
if (method == 'error') {
|
|
final err = triple[1] as Map<String, dynamic>;
|
|
throw JmapException('$expectedMethod error: ${err['type']}');
|
|
}
|
|
return triple[1] as Map<String, dynamic>;
|
|
}
|
|
|
|
String _encodeJmapAddresses(dynamic addressList) {
|
|
if (addressList == null) return '[]';
|
|
final list = addressList as List<dynamic>;
|
|
return jsonEncode(
|
|
list
|
|
.map(
|
|
(a) => {
|
|
'name': (a as Map<String, dynamic>)['name'],
|
|
'email': a['email'],
|
|
},
|
|
)
|
|
.toList(),
|
|
);
|
|
}
|
|
|
|
DateTime? _parseDate(String? iso) =>
|
|
iso == null ? null : DateTime.tryParse(iso);
|
|
|
|
// ── Mutations ──────────────────────────────────────────────────────────────
|
|
|
|
@override
|
|
Future<void> setFlag(
|
|
String emailId, {
|
|
bool? seen,
|
|
bool? flagged,
|
|
}) async {
|
|
final row = await (_db.select(_db.emails)
|
|
..where((t) => t.id.equals(emailId)))
|
|
.getSingle();
|
|
final account = (await _accounts.getAccount(row.accountId))!;
|
|
|
|
if (account.type == account_model.AccountType.jmap) {
|
|
if (seen != null) {
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'flag_seen',
|
|
jsonEncode({'seen': seen}),
|
|
);
|
|
}
|
|
if (flagged != null) {
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'flag_flagged',
|
|
jsonEncode({'flagged': flagged}),
|
|
);
|
|
}
|
|
// Optimistic local update.
|
|
await (_db.update(_db.emails)..where((t) => t.id.equals(emailId))).write(
|
|
EmailsCompanion(
|
|
isSeen: seen != null ? Value(seen) : const Value.absent(),
|
|
isFlagged: flagged != null ? Value(flagged) : const Value.absent(),
|
|
),
|
|
);
|
|
return;
|
|
}
|
|
|
|
if (seen != null) {
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'flag_seen',
|
|
jsonEncode(
|
|
{'uid': row.uid, 'mailboxPath': row.mailboxPath, 'seen': seen},
|
|
),
|
|
);
|
|
}
|
|
if (flagged != null) {
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'flag_flagged',
|
|
jsonEncode({
|
|
'uid': row.uid,
|
|
'mailboxPath': row.mailboxPath,
|
|
'flagged': flagged,
|
|
}),
|
|
);
|
|
}
|
|
await (_db.update(_db.emails)..where((t) => t.id.equals(emailId))).write(
|
|
EmailsCompanion(
|
|
isSeen: seen != null ? Value(seen) : const Value.absent(),
|
|
isFlagged: flagged != null ? Value(flagged) : const Value.absent(),
|
|
),
|
|
);
|
|
}
|
|
|
|
@override
|
|
Future<void> moveEmail(String emailId, String destMailboxPath) async {
|
|
final row = await (_db.select(_db.emails)
|
|
..where((t) => t.id.equals(emailId)))
|
|
.getSingle();
|
|
final account = (await _accounts.getAccount(row.accountId))!;
|
|
|
|
if (account.type == account_model.AccountType.jmap) {
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'move',
|
|
jsonEncode({'dest': destMailboxPath}),
|
|
);
|
|
// Optimistic: remove from current view; next sync will reconcile.
|
|
await (_db.delete(_db.emails)..where((t) => t.id.equals(emailId))).go();
|
|
return;
|
|
}
|
|
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'move',
|
|
jsonEncode({
|
|
'uid': row.uid,
|
|
'mailboxPath': row.mailboxPath,
|
|
'dest': destMailboxPath,
|
|
}),
|
|
);
|
|
await (_db.delete(_db.emails)..where((t) => t.id.equals(emailId))).go();
|
|
}
|
|
|
|
@override
|
|
Future<void> deleteEmail(String emailId) async {
|
|
final row = await (_db.select(_db.emails)
|
|
..where((t) => t.id.equals(emailId)))
|
|
.getSingle();
|
|
final account = (await _accounts.getAccount(row.accountId))!;
|
|
|
|
if (account.type == account_model.AccountType.jmap) {
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'delete',
|
|
jsonEncode(<String, dynamic>{}),
|
|
);
|
|
await (_db.delete(_db.emails)..where((t) => t.id.equals(emailId))).go();
|
|
return;
|
|
}
|
|
|
|
await _enqueueChange(
|
|
account.id,
|
|
emailId,
|
|
'delete',
|
|
jsonEncode({'uid': row.uid, 'mailboxPath': row.mailboxPath}),
|
|
);
|
|
await (_db.delete(_db.emails)..where((t) => t.id.equals(emailId))).go();
|
|
}
|
|
|
|
// ── pending_changes queue ──────────────────────────────────────────────────
|
|
|
|
Future<void> _enqueueChange(
|
|
String accountId,
|
|
String resourceId,
|
|
String changeType,
|
|
String payload,
|
|
) async {
|
|
await _db.into(_db.pendingChanges).insert(
|
|
PendingChangesCompanion.insert(
|
|
accountId: accountId,
|
|
resourceType: 'Email',
|
|
resourceId: resourceId,
|
|
changeType: changeType,
|
|
payload: payload,
|
|
createdAt: DateTime.now(),
|
|
),
|
|
);
|
|
}
|
|
|
|
/// Drains pending changes for [accountId] via the appropriate protocol.
|
|
/// Called at the start of each sync cycle. Returns count of applied changes.
|
|
@override
|
|
Future<int> flushPendingChanges(String accountId, String password) async {
|
|
final rows = await (_db.select(_db.pendingChanges)
|
|
..where((t) => t.accountId.equals(accountId))
|
|
..orderBy([(t) => OrderingTerm.asc(t.createdAt)]))
|
|
.get();
|
|
if (rows.isEmpty) return 0;
|
|
|
|
final account = (await _accounts.getAccount(accountId))!;
|
|
switch (account.type) {
|
|
case account_model.AccountType.imap:
|
|
return _flushPendingChangesImap(account, password, rows);
|
|
case account_model.AccountType.jmap:
|
|
return _flushPendingChangesJmap(account, password, rows);
|
|
}
|
|
}
|
|
|
|
Future<int> _flushPendingChangesJmap(
|
|
account_model.Account account,
|
|
String password,
|
|
List<PendingChangeRow> rows,
|
|
) async {
|
|
final jmapUrl = account.jmapUrl;
|
|
if (jmapUrl == null || jmapUrl.isEmpty) return 0;
|
|
|
|
final jmap = await JmapClient.connect(
|
|
httpClient: _httpClient,
|
|
jmapUrl: Uri.parse(jmapUrl),
|
|
username: _effectiveUsername(account),
|
|
password: password,
|
|
);
|
|
|
|
final ifInState = await _loadSyncState(account.id, 'Email');
|
|
var applied = 0;
|
|
|
|
for (final row in rows) {
|
|
try {
|
|
final newState =
|
|
await _applyPendingChangeJmap(jmap, row, ifInState: ifInState);
|
|
await (_db.delete(_db.pendingChanges)
|
|
..where((t) => t.id.equals(row.id)))
|
|
.go();
|
|
applied++;
|
|
// Keep our checkpoint in sync with whatever the server returned.
|
|
if (newState != null) {
|
|
await _saveSyncState(account.id, 'Email', newState);
|
|
}
|
|
} on JmapStateMismatchException {
|
|
// Server rejected the mutation because our state token is stale.
|
|
// Drop the cached state so the next sync cycle does a full re-fetch,
|
|
// after which this change will be retried with a fresh token.
|
|
await (_db.delete(_db.syncStates)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(account.id) &
|
|
t.resourceType.equals('Email'),
|
|
))
|
|
.go();
|
|
await _recordChangeError(
|
|
row,
|
|
'stateMismatch — will retry after re-sync',
|
|
);
|
|
// State is now stale for all remaining rows too; stop processing.
|
|
break;
|
|
} on JmapSetItemException catch (e) {
|
|
// Permanent per-item rejection (e.g. notFound, forbidden) — discard
|
|
// the change so the queue doesn't grow unboundedly.
|
|
await (_db.delete(_db.pendingChanges)
|
|
..where((t) => t.id.equals(row.id)))
|
|
.go();
|
|
log('JMAP permanent error for change ${row.id}: $e');
|
|
} catch (e) {
|
|
await _recordChangeError(row, e);
|
|
}
|
|
}
|
|
return applied;
|
|
}
|
|
|
|
Future<int> _flushPendingChangesImap(
|
|
account_model.Account account,
|
|
String password,
|
|
List<PendingChangeRow> rows,
|
|
) async {
|
|
imap.ImapClient? client;
|
|
try {
|
|
client =
|
|
await _imapConnect(account, _effectiveUsername(account), password);
|
|
} catch (e) {
|
|
// Connection-level failure — bump all rows, they'll retry next cycle.
|
|
for (final row in rows) {
|
|
await _recordChangeError(row, e);
|
|
}
|
|
return 0;
|
|
}
|
|
var applied = 0;
|
|
try {
|
|
for (final row in rows) {
|
|
try {
|
|
await _applyPendingChangeImap(client, row);
|
|
await (_db.delete(_db.pendingChanges)
|
|
..where((t) => t.id.equals(row.id)))
|
|
.go();
|
|
applied++;
|
|
} catch (e) {
|
|
await _recordChangeError(row, e);
|
|
}
|
|
}
|
|
} finally {
|
|
await client.logout();
|
|
}
|
|
return applied;
|
|
}
|
|
|
|
Future<void> _applyPendingChangeImap(
|
|
imap.ImapClient client,
|
|
PendingChangeRow row,
|
|
) async {
|
|
final payload = jsonDecode(row.payload) as Map<String, dynamic>;
|
|
final uid = payload['uid'] as int;
|
|
final mailboxPath = payload['mailboxPath'] as String;
|
|
final seq = imap.MessageSequence.fromId(uid, isUid: true);
|
|
await client.selectMailboxByPath(mailboxPath);
|
|
|
|
switch (row.changeType) {
|
|
case 'flag_seen':
|
|
final seen = payload['seen'] as bool;
|
|
seen ? await client.uidMarkSeen(seq) : await client.uidMarkUnseen(seq);
|
|
case 'flag_flagged':
|
|
final flagged = payload['flagged'] as bool;
|
|
flagged
|
|
? await client.uidMarkFlagged(seq)
|
|
: await client.uidMarkUnflagged(seq);
|
|
case 'move':
|
|
await client.uidMove(seq, targetMailboxPath: payload['dest'] as String);
|
|
case 'delete':
|
|
await client.uidMarkDeleted(seq);
|
|
await client.uidExpunge(seq);
|
|
}
|
|
}
|
|
|
|
/// Applies a single pending change to the JMAP server.
|
|
///
|
|
/// Returns the `newState` from the server's `Email/set` response so the
|
|
/// caller can keep the local checkpoint in sync.
|
|
///
|
|
/// Throws [JmapStateMismatchException] when the server rejects the request
|
|
/// because [ifInState] is stale (RFC 8620 §5.3 `stateMismatch`).
|
|
Future<String?> _applyPendingChangeJmap(
|
|
JmapClient jmap,
|
|
PendingChangeRow row, {
|
|
String? ifInState,
|
|
}) async {
|
|
final payload = jsonDecode(row.payload) as Map<String, dynamic>;
|
|
// Extract the JMAP email ID from the DB id (format: "accountId:jmapId").
|
|
final jmapEmailId = row.resourceId.contains(':')
|
|
? row.resourceId.substring(row.resourceId.indexOf(':') + 1)
|
|
: row.resourceId;
|
|
|
|
Map<String, dynamic> setArgs(Map<String, dynamic> extra) => {
|
|
'accountId': jmap.accountId,
|
|
if (ifInState != null) 'ifInState': ifInState,
|
|
...extra,
|
|
};
|
|
|
|
List<dynamic> responses;
|
|
switch (row.changeType) {
|
|
case 'flag_seen':
|
|
final seen = payload['seen'] as bool;
|
|
responses = await jmap.call([
|
|
[
|
|
'Email/set',
|
|
setArgs({
|
|
'update': {
|
|
jmapEmailId: {'keywords/\$seen': seen},
|
|
},
|
|
}),
|
|
'0',
|
|
]
|
|
]);
|
|
|
|
case 'flag_flagged':
|
|
final flagged = payload['flagged'] as bool;
|
|
responses = await jmap.call([
|
|
[
|
|
'Email/set',
|
|
setArgs({
|
|
'update': {
|
|
jmapEmailId: {'keywords/\$flagged': flagged},
|
|
},
|
|
}),
|
|
'0',
|
|
]
|
|
]);
|
|
|
|
case 'move':
|
|
final destMailboxId = payload['dest'] as String;
|
|
responses = await jmap.call([
|
|
[
|
|
'Email/set',
|
|
setArgs({
|
|
'update': {
|
|
jmapEmailId: {
|
|
'mailboxIds/$destMailboxId': true,
|
|
'mailboxIds/${row.resourceId}': null,
|
|
},
|
|
},
|
|
}),
|
|
'0',
|
|
]
|
|
]);
|
|
|
|
case 'delete':
|
|
responses = await jmap.call([
|
|
[
|
|
'Email/set',
|
|
setArgs({
|
|
'destroy': [jmapEmailId],
|
|
}),
|
|
'0',
|
|
]
|
|
]);
|
|
|
|
default:
|
|
return null;
|
|
}
|
|
|
|
final result = _responseArgs(responses, 0, 'Email/set');
|
|
|
|
// stateMismatch is returned as a top-level error in the Email/set response
|
|
// (not the per-method error handled by _responseArgs).
|
|
if (result['type'] == 'stateMismatch') {
|
|
throw const JmapStateMismatchException();
|
|
}
|
|
|
|
// Check for per-item rejection (notUpdated / notDestroyed).
|
|
final notUpdated = result['notUpdated'] as Map<String, dynamic>?;
|
|
if (notUpdated != null && notUpdated.containsKey(jmapEmailId)) {
|
|
final err = notUpdated[jmapEmailId] as Map<String, dynamic>;
|
|
throw JmapSetItemException(
|
|
err['type'] as String? ?? 'unknown',
|
|
err['description'] as String?,
|
|
);
|
|
}
|
|
final notDestroyed = result['notDestroyed'] as Map<String, dynamic>?;
|
|
if (notDestroyed != null && notDestroyed.containsKey(jmapEmailId)) {
|
|
final err = notDestroyed[jmapEmailId] as Map<String, dynamic>;
|
|
throw JmapSetItemException(
|
|
err['type'] as String? ?? 'unknown',
|
|
err['description'] as String?,
|
|
);
|
|
}
|
|
|
|
return result['newState'] as String?;
|
|
}
|
|
|
|
@override
|
|
Future<void> sendEmail(String accountId, model.EmailDraft draft) async {
|
|
final account = (await _accounts.getAccount(accountId))!;
|
|
final password = await _accounts.getPassword(accountId);
|
|
switch (account.type) {
|
|
case account_model.AccountType.imap:
|
|
await _sendEmailImap(account, password, draft);
|
|
case account_model.AccountType.jmap:
|
|
await _sendEmailJmap(account, password, draft);
|
|
}
|
|
}
|
|
|
|
Future<void> _sendEmailImap(
|
|
account_model.Account account,
|
|
String password,
|
|
model.EmailDraft draft,
|
|
) async {
|
|
final builder = imap.MessageBuilder()
|
|
..from = [imap.MailAddress(draft.from.name, draft.from.email)]
|
|
..to = draft.to.map((a) => imap.MailAddress(a.name, a.email)).toList()
|
|
..cc = draft.cc.map((a) => imap.MailAddress(a.name, a.email)).toList()
|
|
..subject = draft.subject
|
|
..text = draft.body;
|
|
for (final filePath in draft.attachmentFilePaths) {
|
|
final file = File(filePath);
|
|
final mediaType = imap.MediaType.guessFromFileName(filePath);
|
|
await builder.addFile(file, mediaType);
|
|
}
|
|
final mimeMessage = builder.buildMimeMessage();
|
|
final smtpClient =
|
|
await _smtpConnect(account, _effectiveUsername(account), password);
|
|
try {
|
|
await smtpClient.sendMessage(mimeMessage);
|
|
} finally {
|
|
await smtpClient.quit();
|
|
}
|
|
// Save a copy to the Sent folder via IMAP APPEND.
|
|
// Create the folder first — many servers don't pre-create it.
|
|
final imapClient =
|
|
await _imapConnect(account, _effectiveUsername(account), password);
|
|
try {
|
|
try {
|
|
await imapClient.createMailbox('Sent');
|
|
} catch (_) {
|
|
// Already exists — that's fine.
|
|
}
|
|
await imapClient.appendMessage(
|
|
mimeMessage,
|
|
targetMailboxPath: 'Sent',
|
|
flags: [r'\Seen'],
|
|
);
|
|
} finally {
|
|
await imapClient.logout();
|
|
}
|
|
}
|
|
|
|
Future<void> _sendEmailJmap(
|
|
account_model.Account account,
|
|
String password,
|
|
model.EmailDraft draft,
|
|
) async {
|
|
final jmapUrl = account.jmapUrl;
|
|
if (jmapUrl == null || jmapUrl.isEmpty) {
|
|
throw Exception('JMAP account ${account.id} has no jmapUrl');
|
|
}
|
|
final jmap = await JmapClient.connect(
|
|
httpClient: _httpClient,
|
|
jmapUrl: Uri.parse(jmapUrl),
|
|
username: _effectiveUsername(account),
|
|
password: password,
|
|
);
|
|
|
|
// Upload any file attachments and collect their blobIds.
|
|
final attachments = <Map<String, dynamic>>[];
|
|
for (final filePath in draft.attachmentFilePaths) {
|
|
final file = File(filePath);
|
|
final bytes = await file.readAsBytes();
|
|
final contentType = imap.MediaType.guessFromFileName(filePath).text;
|
|
final blobId = await jmap.uploadBlob(bytes, contentType);
|
|
attachments.add({
|
|
'blobId': blobId,
|
|
'type': contentType,
|
|
'name': p.basename(filePath),
|
|
'size': bytes.length,
|
|
'disposition': 'attachment',
|
|
});
|
|
}
|
|
|
|
// Look up the Sent mailbox JMAP ID from the local DB.
|
|
final sentMailbox = await (_db.select(_db.mailboxes)
|
|
..where((t) => t.accountId.equals(account.id) & t.role.equals('sent'))
|
|
..limit(1))
|
|
.getSingleOrNull();
|
|
final sentJmapId = sentMailbox?.path;
|
|
|
|
// Build the email body.
|
|
const bodyPartId = '1';
|
|
final emailCreate = {
|
|
'from': [
|
|
{'name': draft.from.name, 'email': draft.from.email},
|
|
],
|
|
'to': draft.to.map((a) => {'name': a.name, 'email': a.email}).toList(),
|
|
if (draft.cc.isNotEmpty)
|
|
'cc': draft.cc.map((a) => {'name': a.name, 'email': a.email}).toList(),
|
|
'subject': draft.subject,
|
|
'bodyValues': {
|
|
bodyPartId: {
|
|
'value': draft.body,
|
|
'isEncodingProblem': false,
|
|
'isTruncated': false,
|
|
},
|
|
},
|
|
'textBody': [
|
|
{'partId': bodyPartId, 'type': 'text/plain'},
|
|
],
|
|
if (attachments.isNotEmpty) 'attachments': attachments,
|
|
'keywords': {r'$seen': true},
|
|
if (sentJmapId != null) 'mailboxIds': {sentJmapId: true},
|
|
};
|
|
|
|
// Build the recipient envelope for EmailSubmission.
|
|
final allRecipients = [
|
|
...draft.to.map((a) => {'email': a.email}),
|
|
...draft.cc.map((a) => {'email': a.email}),
|
|
];
|
|
|
|
// Chain Email/set (create) + EmailSubmission/set (create) in one request.
|
|
final responses = await jmap.call(
|
|
[
|
|
[
|
|
'Email/set',
|
|
{
|
|
'accountId': jmap.accountId,
|
|
'create': {'em1': emailCreate},
|
|
},
|
|
'0',
|
|
],
|
|
[
|
|
'EmailSubmission/set',
|
|
{
|
|
'accountId': jmap.accountId,
|
|
'create': {
|
|
'sub1': {
|
|
'#emailId': {
|
|
'resultOf': '0',
|
|
'name': 'Email/set',
|
|
'path': '/created/em1/id',
|
|
},
|
|
'envelope': {
|
|
'mailFrom': {'email': draft.from.email},
|
|
'rcptTo': allRecipients,
|
|
},
|
|
},
|
|
},
|
|
},
|
|
'1',
|
|
],
|
|
],
|
|
withSubmission: true,
|
|
);
|
|
|
|
// Check Email/set for creation errors.
|
|
final setResult = _responseArgs(responses, 0, 'Email/set');
|
|
final notCreated = setResult['notCreated'] as Map<String, dynamic>?;
|
|
if (notCreated != null && notCreated.containsKey('em1')) {
|
|
final err = notCreated['em1'] as Map<String, dynamic>;
|
|
throw JmapException('Email/set create failed: ${err['type']}');
|
|
}
|
|
|
|
// Check EmailSubmission/set for submission errors.
|
|
final subResult = _responseArgs(responses, 1, 'EmailSubmission/set');
|
|
final notSubmitted = subResult['notCreated'] as Map<String, dynamic>?;
|
|
if (notSubmitted != null && notSubmitted.containsKey('sub1')) {
|
|
final err = notSubmitted['sub1'] as Map<String, dynamic>;
|
|
throw JmapException('EmailSubmission/set failed: ${err['type']}');
|
|
}
|
|
}
|
|
|
|
@override
|
|
Future<String> downloadAttachment(
|
|
String emailId,
|
|
model.EmailAttachment attachment,
|
|
) async {
|
|
final cacheDir = await _getCacheDir();
|
|
final dir = Directory(
|
|
p.join(
|
|
cacheDir.path,
|
|
'sharedinbox',
|
|
'attachments',
|
|
emailId.replaceAll(':', '_'),
|
|
),
|
|
);
|
|
await dir.create(recursive: true);
|
|
|
|
final file = File(p.join(dir.path, attachment.filename));
|
|
if (await file.exists()) return file.path;
|
|
|
|
if (attachment.fetchPartId.isEmpty) {
|
|
throw StateError(
|
|
'Cannot download ${attachment.filename}: missing part ID. '
|
|
'Open the email again to refresh.',
|
|
);
|
|
}
|
|
|
|
final emailRow = await (_db.select(_db.emails)
|
|
..where((t) => t.id.equals(emailId)))
|
|
.getSingle();
|
|
final account = (await _accounts.getAccount(emailRow.accountId))!;
|
|
final password = await _accounts.getPassword(account.id);
|
|
|
|
if (account.type == account_model.AccountType.jmap) {
|
|
final jmap = await JmapClient.connect(
|
|
httpClient: _httpClient,
|
|
jmapUrl: Uri.parse(account.jmapUrl!),
|
|
username: _effectiveUsername(account),
|
|
password: password,
|
|
);
|
|
final bytes = await jmap.downloadBlob(
|
|
attachment.fetchPartId,
|
|
name: attachment.filename,
|
|
type: attachment.contentType,
|
|
);
|
|
await file.writeAsBytes(bytes);
|
|
return file.path;
|
|
}
|
|
|
|
final client =
|
|
await _imapConnect(account, _effectiveUsername(account), password);
|
|
try {
|
|
await client.selectMailboxByPath(emailRow.mailboxPath);
|
|
final fetch = await client.uidFetchMessage(
|
|
emailRow.uid,
|
|
'BODY[${attachment.fetchPartId}]',
|
|
);
|
|
final msg = fetch.messages.first;
|
|
final part = msg.getPart(attachment.fetchPartId);
|
|
final bytes = part?.decodeContentBinary();
|
|
if (bytes == null) {
|
|
throw StateError(
|
|
'Failed to decode attachment ${attachment.filename}.',
|
|
);
|
|
}
|
|
await file.writeAsBytes(bytes);
|
|
return file.path;
|
|
} finally {
|
|
await client.logout();
|
|
}
|
|
}
|
|
|
|
@override
|
|
Future<List<model.Email>> searchEmailsGlobal(
|
|
String accountId,
|
|
String query,
|
|
) async {
|
|
final pattern = '%${query.toLowerCase()}%';
|
|
final rows = await (_db.select(_db.emails)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(accountId) &
|
|
(t.subject.like(pattern) | t.preview.like(pattern)),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.desc(t.receivedAt)])
|
|
..limit(50))
|
|
.get();
|
|
return rows.map(_toModel).toList();
|
|
}
|
|
|
|
@override
|
|
Future<List<model.Email>> getEmailsByAddress(
|
|
String accountId,
|
|
String address,
|
|
) async {
|
|
final pattern = '%${address.toLowerCase()}%';
|
|
final rows = await (_db.select(_db.emails)
|
|
..where(
|
|
(t) =>
|
|
t.accountId.equals(accountId) &
|
|
(t.fromJson.like(pattern) |
|
|
t.toAddresses.like(pattern) |
|
|
t.ccJson.like(pattern)),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.desc(t.receivedAt)]))
|
|
.get();
|
|
return rows.map(_toModel).toList();
|
|
}
|
|
|
|
@override
|
|
Future<List<model.Email>> searchEmails(
|
|
String accountId,
|
|
String mailboxPath,
|
|
String query,
|
|
) async {
|
|
final account = (await _accounts.getAccount(accountId))!;
|
|
final password = await _accounts.getPassword(accountId);
|
|
final client =
|
|
await _imapConnect(account, _effectiveUsername(account), password);
|
|
try {
|
|
await client.selectMailboxByPath(mailboxPath);
|
|
final escaped = query.replaceAll('"', '\\"');
|
|
final result = await client.uidSearchMessages(
|
|
searchCriteria: 'OR SUBJECT "$escaped" TEXT "$escaped"',
|
|
);
|
|
final uids = result.matchingSequence?.toList() ?? [];
|
|
if (uids.isEmpty) return [];
|
|
|
|
final fetch = await client.uidFetchMessages(
|
|
imap.MessageSequence.fromIds(uids, isUid: true),
|
|
'(UID FLAGS ENVELOPE)',
|
|
);
|
|
return fetch.messages
|
|
.where((msg) => msg.uid != null && msg.envelope != null)
|
|
.map((msg) {
|
|
final envelope = msg.envelope!;
|
|
final uid = msg.uid!;
|
|
final emailId = '$accountId:$uid';
|
|
return model.Email(
|
|
id: emailId,
|
|
accountId: accountId,
|
|
mailboxPath: mailboxPath,
|
|
uid: uid,
|
|
subject: envelope.subject,
|
|
sentAt: envelope.date,
|
|
receivedAt: envelope.date ?? DateTime.now(),
|
|
from: _toAddressList(envelope.from),
|
|
to: _toAddressList(envelope.to),
|
|
cc: _toAddressList(envelope.cc),
|
|
isSeen: msg.flags?.contains(r'\Seen') ?? false,
|
|
isFlagged: msg.flags?.contains(r'\Flagged') ?? false,
|
|
hasAttachment: msg.hasAttachments(),
|
|
);
|
|
}).toList();
|
|
} finally {
|
|
await client.logout();
|
|
}
|
|
}
|
|
|
|
List<model.EmailAddress> _toAddressList(List<imap.MailAddress>? addresses) =>
|
|
(addresses ?? const [])
|
|
.map(
|
|
(a) => model.EmailAddress(
|
|
name: a.personalName,
|
|
email: a.email,
|
|
),
|
|
)
|
|
.toList();
|
|
|
|
// ── Helpers ────────────────────────────────────────────────────────────────
|
|
|
|
String _encodeAddresses(List<imap.MailAddress>? addresses) => jsonEncode(
|
|
(addresses ?? const [])
|
|
.map((a) => {'name': a.personalName, 'email': a.email})
|
|
.toList(),
|
|
);
|
|
|
|
model.Email _toModel(Email row) {
|
|
List<model.EmailAddress> parseAddresses(String json) {
|
|
final list = jsonDecode(json) as List<dynamic>;
|
|
return list
|
|
.map(
|
|
(e) => model.EmailAddress(
|
|
name: (e as Map<String, dynamic>)['name'] as String?,
|
|
email: e['email'] as String,
|
|
),
|
|
)
|
|
.toList();
|
|
}
|
|
|
|
return model.Email(
|
|
id: row.id,
|
|
accountId: row.accountId,
|
|
mailboxPath: row.mailboxPath,
|
|
uid: row.uid,
|
|
subject: row.subject,
|
|
sentAt: row.sentAt,
|
|
receivedAt: row.receivedAt,
|
|
from: parseAddresses(row.fromJson),
|
|
to: parseAddresses(row.toAddresses),
|
|
cc: parseAddresses(row.ccJson),
|
|
preview: row.preview,
|
|
isSeen: row.isSeen,
|
|
isFlagged: row.isFlagged,
|
|
hasAttachment: row.hasAttachment,
|
|
);
|
|
}
|
|
|
|
model.EmailBody _bodyRowToModel(EmailBody row) => model.EmailBody(
|
|
emailId: row.emailId,
|
|
textBody: row.textBody,
|
|
htmlBody: row.htmlBody,
|
|
attachments: _parseAttachments(row.attachmentsJson),
|
|
);
|
|
|
|
List<model.EmailAttachment> _parseAttachments(String json) {
|
|
final list = jsonDecode(json) as List<dynamic>;
|
|
return list
|
|
.map(
|
|
(e) => model.EmailAttachment(
|
|
filename: (e as Map<String, dynamic>)['filename'] as String,
|
|
contentType: e['contentType'] as String,
|
|
size: e['size'] as int,
|
|
fetchPartId: (e['fetchPartId'] as String?) ?? '',
|
|
),
|
|
)
|
|
.toList();
|
|
}
|
|
|
|
// ── Failed mutations (offline compose queue) ─────────────────────────────
|
|
|
|
@override
|
|
Stream<List<model.FailedMutation>> observeFailedMutations(String accountId) {
|
|
return (_db.select(_db.pendingChanges)
|
|
..where(
|
|
(t) => t.accountId.equals(accountId) & t.lastError.isNotNull(),
|
|
)
|
|
..orderBy([(t) => OrderingTerm.asc(t.createdAt)]))
|
|
.watch()
|
|
.map(
|
|
(rows) => rows
|
|
.map(
|
|
(r) => model.FailedMutation(
|
|
id: r.id,
|
|
accountId: r.accountId,
|
|
changeType: r.changeType,
|
|
resourceId: r.resourceId,
|
|
lastError: r.lastError!,
|
|
attempts: r.attempts,
|
|
createdAt: r.createdAt,
|
|
),
|
|
)
|
|
.toList(),
|
|
);
|
|
}
|
|
|
|
@override
|
|
Future<void> discardMutation(int id) async {
|
|
await (_db.delete(_db.pendingChanges)..where((t) => t.id.equals(id))).go();
|
|
}
|
|
|
|
@override
|
|
Future<void> retryMutation(int id) async {
|
|
await (_db.update(_db.pendingChanges)..where((t) => t.id.equals(id))).write(
|
|
const PendingChangesCompanion(
|
|
attempts: Value(0),
|
|
lastError: Value(null),
|
|
),
|
|
);
|
|
}
|
|
}
|