FossilRepo

fossilrepo / fossil / tasks.py
Blame History Raw 592 lines
1
"""Celery tasks for Fossil repository management."""
2
3
import hashlib
4
import logging
5
6
from celery import shared_task
7
from django.core.files.base import ContentFile
8
9
logger = logging.getLogger(__name__)
10
11
12
@shared_task(name="fossil.sync_metadata")
13
def sync_repository_metadata():
14
"""Update metadata for all FossilRepository records from disk."""
15
from fossil.models import FossilRepository
16
from fossil.reader import FossilReader
17
18
for repo in FossilRepository.objects.all():
19
if not repo.exists_on_disk:
20
continue
21
try:
22
repo.file_size_bytes = repo.full_path.stat().st_size
23
with FossilReader(repo.full_path) as reader:
24
repo.checkin_count = reader.get_checkin_count()
25
timeline = reader.get_timeline(limit=1)
26
if timeline:
27
repo.last_checkin_at = timeline[0].timestamp
28
repo.fossil_project_code = reader.get_project_code()
29
repo.save(update_fields=["file_size_bytes", "checkin_count", "last_checkin_at", "fossil_project_code", "updated_at", "version"])
30
except Exception:
31
logger.exception("Failed to sync metadata for %s", repo.filename)
32
33
34
@shared_task(name="fossil.create_snapshot")
35
def create_snapshot(repository_id: int, note: str = ""):
36
"""Create a FossilSnapshot if FOSSIL_STORE_IN_DB is enabled."""
37
from constance import config
38
39
if not config.FOSSIL_STORE_IN_DB:
40
return
41
42
from fossil.models import FossilRepository, FossilSnapshot
43
44
try:
45
repo = FossilRepository.objects.get(pk=repository_id)
46
except FossilRepository.DoesNotExist:
47
return
48
49
if not repo.exists_on_disk:
50
return
51
52
data = repo.full_path.read_bytes()
53
sha = hashlib.sha256(data).hexdigest()
54
55
# Skip if latest snapshot has same hash
56
latest = repo.snapshots.first()
57
if latest and latest.fossil_hash == sha:
58
return
59
60
snapshot = FossilSnapshot(
61
repository=repo,
62
file_size_bytes=len(data),
63
fossil_hash=sha,
64
note=note,
65
created_by=repo.created_by,
66
)
67
snapshot.file.save(f"{repo.filename}_{sha[:8]}.fossil", ContentFile(data), save=True)
68
logger.info("Created snapshot for %s (hash: %s)", repo.filename, sha[:8])
69
70
71
@shared_task(name="fossil.check_upstream")
72
def check_upstream_updates():
73
"""Check all repos with remote URLs for available updates."""
74
from fossil.cli import FossilCLI
75
from fossil.models import FossilRepository
76
77
cli = FossilCLI()
78
if not cli.is_available():
79
return
80
81
from django.utils import timezone
82
83
for repo in FossilRepository.objects.exclude(remote_url=""):
84
if not repo.exists_on_disk:
85
continue
86
try:
87
result = cli.pull(repo.full_path)
88
if result["success"] and result["artifacts_received"] > 0:
89
repo.upstream_artifacts_available = result["artifacts_received"]
90
repo.last_sync_at = timezone.now()
91
# Update metadata after pull
92
from fossil.reader import FossilReader
93
94
with FossilReader(repo.full_path) as reader:
95
repo.checkin_count = reader.get_checkin_count()
96
timeline = reader.get_timeline(limit=1, event_type="ci")
97
if timeline:
98
repo.last_checkin_at = timeline[0].timestamp
99
repo.file_size_bytes = repo.full_path.stat().st_size
100
repo.save(
101
update_fields=[
102
"upstream_artifacts_available",
103
"last_sync_at",
104
"checkin_count",
105
"last_checkin_at",
106
"file_size_bytes",
107
"updated_at",
108
"version",
109
]
110
)
111
logger.info("Pulled %d artifacts for %s (new count: %d)", result["artifacts_received"], repo.filename, repo.checkin_count)
112
else:
113
repo.upstream_artifacts_available = 0
114
repo.last_sync_at = timezone.now()
115
repo.save(update_fields=["upstream_artifacts_available", "last_sync_at", "updated_at", "version"])
116
except Exception:
117
logger.exception("Failed to check upstream for %s", repo.filename)
118
119
120
@shared_task(name="fossil.git_sync")
121
def run_git_sync(mirror_id: int | None = None):
122
"""Run Git export/push for configured mirrors."""
123
from pathlib import Path
124
125
from constance import config
126
from django.utils import timezone
127
128
from fossil.cli import FossilCLI
129
from fossil.sync_models import GitMirror, SyncLog
130
131
cli = FossilCLI()
132
if not cli.is_available():
133
return
134
135
mirrors = GitMirror.objects.filter(deleted_at__isnull=True).exclude(sync_mode="disabled")
136
if mirror_id:
137
mirrors = mirrors.filter(pk=mirror_id)
138
139
mirror_dir = Path(config.GIT_MIRROR_DIR)
140
141
for mirror in mirrors:
142
repo = mirror.repository
143
if not repo.exists_on_disk:
144
continue
145
146
log = SyncLog.objects.create(mirror=mirror, triggered_by="schedule" if not mirror_id else "manual")
147
148
try:
149
# Ensure default user
150
cli.ensure_default_user(repo.full_path)
151
152
# Git export directory for this mirror
153
export_dir = mirror_dir / f"{repo.filename.replace('.fossil', '')}-git"
154
155
# Pass clean URL and token separately -- never embed credentials in URLs
156
push_url = mirror.git_remote_url
157
auth_token = ""
158
if mirror.auth_method == "token" and mirror.auth_credential and push_url.startswith("https://"):
159
auth_token = mirror.auth_credential
160
161
result = cli.git_export(repo.full_path, export_dir, autopush_url=push_url, auth_token=auth_token)
162
163
# Scrub any credential from output before persisting
164
message = result.get("message", "")
165
if mirror.auth_credential:
166
message = message.replace(mirror.auth_credential, "[REDACTED]")
167
168
log.status = "success" if result["success"] else "failed"
169
log.message = message
170
log.completed_at = timezone.now()
171
log.save()
172
173
mirror.last_sync_at = timezone.now()
174
mirror.last_sync_status = log.status
175
mirror.last_sync_message = message[:500]
176
mirror.total_syncs += 1
177
mirror.save(
178
update_fields=[
179
"last_sync_at",
180
"last_sync_status",
181
"last_sync_message",
182
"total_syncs",
183
"updated_at",
184
"version",
185
]
186
)
187
188
if result["success"]:
189
logger.info("Git sync success for %s → %s", repo.filename, mirror.git_remote_url)
190
191
# Chain ticket and wiki sync if enabled
192
if mirror.sync_tickets:
193
sync_tickets_to_github.delay(mirror.id)
194
if mirror.sync_wiki:
195
sync_wiki_to_github.delay(mirror.id)
196
else:
197
logger.warning("Git sync failed for %s: %s", repo.filename, result["message"][:200])
198
199
except Exception:
200
logger.exception("Git sync error for %s", repo.filename)
201
log.status = "failed"
202
log.message = "Unexpected error"
203
log.completed_at = timezone.now()
204
log.save()
205
206
207
@shared_task(name="fossil.send_digest")
208
def send_digest(mode="daily"):
209
"""Send digest emails to users who prefer batch delivery.
210
211
Collects unread notifications for users with the given delivery mode
212
and sends a single summary email with HTML template. Marks those
213
notifications as read after sending.
214
"""
215
from django.conf import settings
216
from django.core.mail import send_mail
217
from django.template.loader import render_to_string
218
219
from fossil.notifications import Notification, NotificationPreference
220
221
prefs = NotificationPreference.objects.filter(delivery_mode=mode).select_related("user")
222
for pref in prefs:
223
unread = Notification.objects.filter(user=pref.user, read=False).select_related("project")
224
if not unread.exists():
225
continue
226
227
count = unread.count()
228
notifications_list = list(unread[:50])
229
overflow_count = count - 50 if count > 50 else 0
230
231
# Plain text fallback
232
lines = [f"You have {count} new notification{'s' if count != 1 else ''}:\n"]
233
for notif in notifications_list:
234
lines.append(f"- [{notif.event_type}] {notif.project.name}: {notif.title}")
235
if overflow_count:
236
lines.append(f"\n... and {overflow_count} more.")
237
238
# HTML version
239
html_body = render_to_string(
240
"email/digest.html",
241
{
242
"digest_type": mode,
243
"count": count,
244
"notifications": notifications_list,
245
"overflow_count": overflow_count,
246
"dashboard_url": "/",
247
"preferences_url": "/auth/notifications/",
248
},
249
)
250
251
try:
252
send_mail(
253
subject=f"Fossilrepo {mode.title()} Digest - {count} update{'s' if count != 1 else ''}",
254
message="\n".join(lines),
255
from_email=settings.DEFAULT_FROM_EMAIL,
256
recipient_list=[pref.user.email],
257
html_message=html_body,
258
fail_silently=True,
259
)
260
except Exception:
261
logger.exception("Failed to send %s digest to %s", mode, pref.user.email)
262
continue
263
264
unread.update(read=True)
265
266
267
@shared_task(name="fossil.dispatch_webhook", bind=True, max_retries=3)
268
def dispatch_webhook(self, webhook_id, event_type, payload):
269
"""Deliver a webhook with retry and logging."""
270
import hashlib
271
import hmac
272
import json
273
import time
274
275
import requests
276
277
from fossil.webhooks import Webhook, WebhookDelivery
278
279
try:
280
webhook = Webhook.objects.get(id=webhook_id)
281
except Webhook.DoesNotExist:
282
logger.warning("Webhook %s not found, skipping delivery", webhook_id)
283
return
284
285
# Re-validate URL at dispatch time (hostname could resolve differently than at save time)
286
from core.url_validation import is_safe_outbound_url
287
288
is_safe, url_error = is_safe_outbound_url(webhook.url)
289
if not is_safe:
290
logger.warning("Webhook %s URL failed safety check at dispatch: %s", webhook_id, url_error)
291
WebhookDelivery.objects.create(
292
webhook=webhook,
293
event_type=event_type,
294
payload=payload,
295
response_status=0,
296
response_body=f"Blocked: {url_error}",
297
success=False,
298
duration_ms=0,
299
attempt=self.request.retries + 1,
300
)
301
return
302
303
headers = {"Content-Type": "application/json", "X-Fossilrepo-Event": event_type}
304
body = json.dumps(payload)
305
306
if webhook.secret:
307
sig = hmac.new(webhook.secret.encode(), body.encode(), hashlib.sha256).hexdigest()
308
headers["X-Fossilrepo-Signature"] = f"sha256={sig}"
309
310
start = time.monotonic()
311
try:
312
resp = requests.post(webhook.url, data=body, headers=headers, timeout=30, allow_redirects=False)
313
duration = int((time.monotonic() - start) * 1000)
314
315
WebhookDelivery.objects.create(
316
webhook=webhook,
317
event_type=event_type,
318
payload=payload,
319
response_status=resp.status_code,
320
response_body=resp.text[:5000],
321
success=200 <= resp.status_code < 300,
322
duration_ms=duration,
323
attempt=self.request.retries + 1,
324
)
325
326
if not (200 <= resp.status_code < 300):
327
raise self.retry(countdown=60 * (2**self.request.retries))
328
except requests.RequestException as exc:
329
duration = int((time.monotonic() - start) * 1000)
330
WebhookDelivery.objects.create(
331
webhook=webhook,
332
event_type=event_type,
333
payload=payload,
334
response_status=0,
335
response_body=str(exc)[:5000],
336
success=False,
337
duration_ms=duration,
338
attempt=self.request.retries + 1,
339
)
340
raise self.retry(exc=exc, countdown=60 * (2**self.request.retries)) from exc
341
342
343
@shared_task(name="fossil.dispatch_notifications")
344
def dispatch_notifications():
345
"""Check for new Fossil events and send notifications to watchers."""
346
import datetime
347
348
from django.utils import timezone
349
350
from fossil.models import FossilRepository
351
from fossil.notifications import ProjectWatch, notify_project_event
352
from fossil.reader import FossilReader
353
354
watched_project_ids = set(ProjectWatch.objects.filter(deleted_at__isnull=True).values_list("project_id", flat=True))
355
if not watched_project_ids:
356
return
357
358
cutoff = timezone.now() - datetime.timedelta(minutes=5)
359
360
for repo in FossilRepository.objects.filter(project_id__in=watched_project_ids, deleted_at__isnull=True):
361
if not repo.exists_on_disk:
362
continue
363
try:
364
with FossilReader(repo.full_path) as reader:
365
entries = reader.get_timeline(limit=10)
366
for entry in entries:
367
if entry.timestamp < cutoff:
368
break
369
event_type = {"ci": "checkin", "w": "wiki", "t": "ticket", "f": "forum"}.get(entry.event_type, "other")
370
url = f"/projects/{repo.project.slug}/fossil/"
371
if entry.event_type == "ci":
372
url += f"checkin/{entry.uuid}/"
373
notify_project_event(
374
project=repo.project,
375
event_type=event_type,
376
title=f"{entry.user}: {entry.comment[:100]}" if entry.comment else f"New {event_type}",
377
body=entry.comment or "",
378
url=url,
379
)
380
except Exception:
381
logger.exception("Notification dispatch error for %s", repo.filename)
382
383
384
@shared_task(name="fossil.sync_tickets_to_github")
385
def sync_tickets_to_github(mirror_id: int):
386
"""Sync Fossil tickets to GitHub Issues for a given mirror.
387
388
Processes up to 20 tickets per run. Resumes from where it left off
389
by checking TicketSyncMapping for already-synced tickets.
390
"""
391
from django.utils import timezone
392
393
from fossil.github_api import GitHubClient, format_ticket_body, fossil_status_to_github, parse_github_repo
394
from fossil.reader import FossilReader
395
from fossil.sync_models import GitMirror, SyncLog, TicketSyncMapping
396
397
try:
398
mirror = GitMirror.objects.get(pk=mirror_id, deleted_at__isnull=True)
399
except GitMirror.DoesNotExist:
400
return
401
402
repo = mirror.repository
403
if not repo.exists_on_disk:
404
return
405
406
parsed = parse_github_repo(mirror.git_remote_url)
407
if not parsed:
408
logger.warning("Cannot parse GitHub owner/repo from %s", mirror.git_remote_url)
409
return
410
411
owner, repo_name = parsed
412
token = mirror.auth_credential
413
if not token:
414
logger.warning("No auth token for mirror %s, skipping ticket sync", mirror_id)
415
return
416
417
log = SyncLog.objects.create(mirror=mirror, triggered_by="ticket_sync")
418
client = GitHubClient(token)
419
synced_count = 0
420
errors = []
421
422
try:
423
with FossilReader(repo.full_path) as reader:
424
tickets = reader.get_tickets(limit=500)
425
426
# Get existing mappings for this mirror
427
existing = {m.fossil_ticket_uuid: m for m in TicketSyncMapping.objects.filter(mirror=mirror)}
428
429
processed = 0
430
for ticket in tickets:
431
if processed >= 20:
432
break
433
434
mapping = existing.get(ticket.uuid)
435
436
# Skip if already synced and status hasn't changed
437
if mapping and mapping.fossil_status == ticket.status:
438
continue
439
440
processed += 1
441
442
# Get full ticket detail + comments
443
with FossilReader(repo.full_path) as reader:
444
detail = reader.get_ticket_detail(ticket.uuid)
445
comments = reader.get_ticket_comments(ticket.uuid)
446
447
if not detail:
448
continue
449
450
gh_state = fossil_status_to_github(detail.status)
451
body = format_ticket_body(detail, comments)
452
453
if mapping:
454
# Update existing issue
455
result = client.update_issue(owner, repo_name, mapping.github_issue_number, title=detail.title, body=body, state=gh_state)
456
if result["success"]:
457
mapping.fossil_status = detail.status
458
mapping.save()
459
synced_count += 1
460
logger.info("Updated GitHub issue #%d for ticket %s", mapping.github_issue_number, ticket.uuid[:10])
461
else:
462
errors.append(f"Update #{mapping.github_issue_number}: {result['error']}")
463
else:
464
# Create new issue
465
result = client.create_issue(owner, repo_name, detail.title, body, state=gh_state)
466
if result["number"]:
467
TicketSyncMapping.objects.create(
468
mirror=mirror,
469
fossil_ticket_uuid=ticket.uuid,
470
github_issue_number=result["number"],
471
fossil_status=detail.status,
472
)
473
synced_count += 1
474
logger.info("Created GitHub issue #%d for ticket %s", result["number"], ticket.uuid[:10])
475
else:
476
errors.append(f"Create '{detail.title[:30]}': {result['error']}")
477
478
log.status = "success" if not errors else "failed"
479
log.artifacts_synced = synced_count
480
log.message = f"Synced {synced_count} tickets." + (f" Errors: {'; '.join(errors[:5])}" if errors else "")
481
log.completed_at = timezone.now()
482
log.save()
483
484
except Exception:
485
logger.exception("Ticket sync error for mirror %s", mirror_id)
486
log.status = "failed"
487
log.message = "Unexpected error during ticket sync"
488
log.completed_at = timezone.now()
489
log.save()
490
491
492
@shared_task(name="fossil.sync_wiki_to_github")
493
def sync_wiki_to_github(mirror_id: int):
494
"""Sync Fossil wiki pages to a wiki/ directory in the GitHub repo.
495
496
Uses the GitHub Contents API to create/update markdown files.
497
Only syncs pages whose content has changed (hash comparison).
498
"""
499
from django.utils import timezone
500
501
from fossil.github_api import GitHubClient, content_hash, parse_github_repo
502
from fossil.reader import FossilReader
503
from fossil.sync_models import GitMirror, SyncLog, WikiSyncMapping
504
505
try:
506
mirror = GitMirror.objects.get(pk=mirror_id, deleted_at__isnull=True)
507
except GitMirror.DoesNotExist:
508
return
509
510
repo = mirror.repository
511
if not repo.exists_on_disk:
512
return
513
514
parsed = parse_github_repo(mirror.git_remote_url)
515
if not parsed:
516
logger.warning("Cannot parse GitHub owner/repo from %s", mirror.git_remote_url)
517
return
518
519
owner, repo_name = parsed
520
token = mirror.auth_credential
521
if not token:
522
logger.warning("No auth token for mirror %s, skipping wiki sync", mirror_id)
523
return
524
525
log = SyncLog.objects.create(mirror=mirror, triggered_by="wiki_sync")
526
client = GitHubClient(token)
527
synced_count = 0
528
errors = []
529
530
try:
531
with FossilReader(repo.full_path) as reader:
532
pages = reader.get_wiki_pages()
533
534
existing = {m.fossil_page_name: m for m in WikiSyncMapping.objects.filter(mirror=mirror)}
535
536
for page in pages:
537
# Read full page content
538
with FossilReader(repo.full_path) as reader:
539
full_page = reader.get_wiki_page(page.name)
540
541
if not full_page or not full_page.content.strip():
542
continue
543
544
current_hash = content_hash(full_page.content)
545
mapping = existing.get(page.name)
546
547
# Skip if content hasn't changed
548
if mapping and mapping.content_hash == current_hash:
549
continue
550
551
# Sanitize page name for file path
552
safe_name = page.name.replace(" ", "-").replace("/", "-")
553
file_path = f"wiki/{safe_name}.md"
554
555
result = client.create_or_update_file(
556
owner,
557
repo_name,
558
file_path,
559
full_page.content,
560
f"Sync wiki page: {page.name}",
561
)
562
563
if result["success"]:
564
if mapping:
565
mapping.content_hash = current_hash
566
mapping.github_path = file_path
567
mapping.save()
568
else:
569
WikiSyncMapping.objects.create(
570
mirror=mirror,
571
fossil_page_name=page.name,
572
content_hash=current_hash,
573
github_path=file_path,
574
)
575
synced_count += 1
576
logger.info("Synced wiki page '%s' to %s", page.name, file_path)
577
else:
578
errors.append(f"'{page.name}': {result['error']}")
579
580
log.status = "success" if not errors else "failed"
581
log.artifacts_synced = synced_count
582
log.message = f"Synced {synced_count} wiki pages." + (f" Errors: {'; '.join(errors[:5])}" if errors else "")
583
log.completed_at = timezone.now()
584
log.save()
585
586
except Exception:
587
logger.exception("Wiki sync error for mirror %s", mirror_id)
588
log.status = "failed"
589
log.message = "Unexpected error during wiki sync"
590
log.completed_at = timezone.now()
591
log.save()
592

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button