#!/usr/bin/env python3 """ FIFO Event Queue Manager - CLI와 모듈 import 양쪽으로 사용 가능 - fcntl.flock 기반 파일 잠금으로 동시 쓰기 충돌 방지 - Atomic write (임시 파일 → os.replace) - JSON 읽기/쓰기 실패 시 .bak 백업 복구 로직 """ import argparse import fcntl import json import os import shutil import sys import tempfile from datetime import datetime, timezone from typing import Optional DEFAULT_QUEUE_FILE = "/home/jay/workspace/memory/events/event-queue.json" class EventQueue: def __init__(self, queue_file: str = None): self.queue_file = queue_file or DEFAULT_QUEUE_FILE self.backup_file = self.queue_file + ".bak" self._ensure_queue_file() # ------------------------------------------------------------------ # 내부 유틸리티 # ------------------------------------------------------------------ def _ensure_queue_file(self): """큐 파일이 없으면 초기 구조로 생성한다.""" os.makedirs(os.path.dirname(self.queue_file), exist_ok=True) if not os.path.exists(self.queue_file): self._write_data({"queue": [], "processed": []}) def _read_data(self) -> dict: """JSON 파일을 읽는다. 실패 시 백업에서 복구를 시도한다.""" try: with open(self.queue_file, "r", encoding="utf-8") as f: fcntl.flock(f, fcntl.LOCK_SH) try: content = f.read() data = json.loads(content) finally: fcntl.flock(f, fcntl.LOCK_UN) # 필수 키 보장 data.setdefault("queue", []) data.setdefault("processed", []) return data except (json.JSONDecodeError, OSError) as primary_err: # 백업 복구 시도 if os.path.exists(self.backup_file): try: with open(self.backup_file, "r", encoding="utf-8") as bf: data = json.load(bf) data.setdefault("queue", []) data.setdefault("processed", []) # 복구된 데이터를 원본 위치에 저장 self._write_data(data) return data except (json.JSONDecodeError, OSError): pass raise RuntimeError( f"큐 파일 읽기 실패: {primary_err}. 백업도 사용 불가." ) def _write_data(self, data: dict): """Atomic write: 임시 파일에 쓴 뒤 os.replace로 교체한다.""" dir_path = os.path.dirname(self.queue_file) os.makedirs(dir_path, exist_ok=True) # 쓰기 전 백업 생성 if os.path.exists(self.queue_file): try: shutil.copy2(self.queue_file, self.backup_file) except OSError: pass # 백업 실패는 무시하고 진행 # 임시 파일에 먼저 쓰기 fd, tmp_path = tempfile.mkstemp(dir=dir_path, suffix=".tmp") try: with os.fdopen(fd, "w", encoding="utf-8") as tf: fcntl.flock(tf, fcntl.LOCK_EX) try: json.dump(data, tf, ensure_ascii=False, indent=2) tf.flush() os.fsync(tf.fileno()) finally: fcntl.flock(tf, fcntl.LOCK_UN) # Atomic rename os.replace(tmp_path, self.queue_file) except Exception: # 임시 파일 정리 try: os.unlink(tmp_path) except OSError: pass raise def _next_event_id(self, data: dict) -> str: """queue + processed 전체에서 최대 시퀀스 번호를 찾아 +1 반환.""" max_seq = 0 all_events = data.get("queue", []) + data.get("processed", []) for evt in all_events: eid = evt.get("id", "") if eid.startswith("evt-"): try: seq = int(eid[4:]) if seq > max_seq: max_seq = seq except ValueError: pass return f"evt-{max_seq + 1:03d}" # ------------------------------------------------------------------ # 공개 API # ------------------------------------------------------------------ def enqueue( self, event_type: str, task_id: str, team_id: str, report_path: str, ) -> dict: """이벤트를 큐에 추가하고 추가된 이벤트 dict를 반환한다.""" # 배타적 잠금을 위해 파일을 열고 read → modify → write os.makedirs(os.path.dirname(self.queue_file), exist_ok=True) lock_path = self.queue_file + ".lock" with open(lock_path, "w", encoding="utf-8") as lock_f: fcntl.flock(lock_f, fcntl.LOCK_EX) try: data = self._read_data() event_id = self._next_event_id(data) event = { "id": event_id, "type": event_type, "task_id": task_id, "team": team_id, "report": report_path, "status": "pending", "created_at": datetime.now(timezone.utc).strftime( "%Y-%m-%dT%H:%M:%S+00:00" ), } data["queue"].append(event) self._write_data(data) finally: fcntl.flock(lock_f, fcntl.LOCK_UN) return event def peek(self) -> Optional[dict]: """pending 상태인 첫 번째 이벤트를 반환한다. 없으면 None.""" data = self._read_data() for evt in data["queue"]: if evt.get("status") == "pending": return evt return None def dequeue(self, event_id: str) -> Optional[dict]: """event_id 이벤트를 queue에서 제거하고 processed로 이동한다.""" lock_path = self.queue_file + ".lock" with open(lock_path, "w", encoding="utf-8") as lock_f: fcntl.flock(lock_f, fcntl.LOCK_EX) try: data = self._read_data() target = None new_queue = [] for evt in data["queue"]: if evt.get("id") == event_id: target = evt else: new_queue.append(evt) if target is None: return None target["status"] = "processed" target["processed_at"] = datetime.now(timezone.utc).strftime( "%Y-%m-%dT%H:%M:%S+00:00" ) data["queue"] = new_queue data["processed"].append(target) self._write_data(data) finally: fcntl.flock(lock_f, fcntl.LOCK_UN) return target def list_events(self, include_processed: bool = False) -> dict: """이벤트 목록을 반환한다.""" data = self._read_data() if include_processed: return { "queue": data["queue"], "processed": data["processed"], } return {"queue": data["queue"]} def count(self) -> int: """pending 이벤트 수를 반환한다.""" data = self._read_data() return sum( 1 for evt in data["queue"] if evt.get("status") == "pending" ) # ------------------------------------------------------------------ # CLI # ------------------------------------------------------------------ def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="event-queue.py", description="FIFO Event Queue Manager", ) subparsers = parser.add_subparsers(dest="command", metavar="COMMAND") # enqueue p_enq = subparsers.add_parser("enqueue", help="이벤트를 큐에 추가한다") p_enq.add_argument("--type", dest="event_type", required=True, help="이벤트 타입 (예: task_complete)") p_enq.add_argument("--task-id", required=True, help="태스크 ID (예: task-42)") p_enq.add_argument("--team", required=True, help="팀 ID (예: dev1-team)") p_enq.add_argument("--report", required=True, help="리포트 파일 경로") p_enq.add_argument("--queue-file", default=None, help="큐 파일 경로 (기본값 사용 가능)") # peek p_peek = subparsers.add_parser("peek", help="다음 처리할 이벤트를 조회한다 (제거 안 함)") p_peek.add_argument("--queue-file", default=None) # dequeue p_deq = subparsers.add_parser("dequeue", help="이벤트를 processed로 이동한다") p_deq.add_argument("event_id", help="이벤트 ID (예: evt-001)") p_deq.add_argument("--queue-file", default=None) # list p_list = subparsers.add_parser("list", help="큐 상태를 조회한다") group = p_list.add_mutually_exclusive_group() group.add_argument("--pending", action="store_true", default=True, help="pending 이벤트만 (기본값)") group.add_argument("--all", dest="all_events", action="store_true", help="queue + processed 모두") p_list.add_argument("--queue-file", default=None) # count p_cnt = subparsers.add_parser("count", help="pending 이벤트 수를 출력한다") p_cnt.add_argument("--queue-file", default=None) return parser def main(): parser = build_parser() args = parser.parse_args() if args.command is None: parser.print_help() sys.exit(0) queue_file = getattr(args, "queue_file", None) eq = EventQueue(queue_file=queue_file) if args.command == "enqueue": result = eq.enqueue( event_type=args.event_type, task_id=args.task_id, team_id=args.team, report_path=args.report, ) print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.command == "peek": result = eq.peek() if result is None: print(json.dumps({"status": "empty"}, ensure_ascii=False, indent=2)) else: print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.command == "dequeue": result = eq.dequeue(args.event_id) if result is None: print( json.dumps( {"error": f"이벤트를 찾을 수 없습니다: {args.event_id}"}, ensure_ascii=False, indent=2, ) ) sys.exit(1) else: print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.command == "list": include_processed = getattr(args, "all_events", False) result = eq.list_events(include_processed=include_processed) print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.command == "count": print(eq.count()) else: parser.print_help() sys.exit(1) if __name__ == "__main__": main()