diff --git a/.gitignore b/.gitignore index 43d7664..5b8aa79 100644 --- a/.gitignore +++ b/.gitignore @@ -51,3 +51,4 @@ out/ # application secrets application-local.yml +.venv \ No newline at end of file diff --git a/README.md b/README.md index e0c1c47..7b6165f 100644 --- a/README.md +++ b/README.md @@ -1,183 +1,445 @@ -# BackEnd -Repository of JobDri BackEnd +# JobDri SERVER -## Docker +**AI 기반 채용 공고 정리 · 모의 지원서 생성 · 자소서 분석 플랫폼 `JobDri`의 백엔드 서버**입니다. -로컬 실행: +> **"채용 공고를 정리하고, 지원서를 만들고, 자소서까지 분석받자."** +> 사용자가 업로드하거나 입력한 채용 공고를 AI가 구조화하고, +> 직무/회사 기반의 **모의 공고와 문항**을 생성하며, +> 작성한 자소서를 **AI 분석 + 실시간 상태 알림**으로 제공하는 **AI 취업 지원 플랫폼**입니다. -```bash -cp .env.example .env -docker compose up --build -``` +Server: `https://api.jobdri.site` -포함 서비스: +## 📖 프로젝트 소개 -- Spring Boot API: `http://localhost:8080` -- FastAPI worker health: `http://localhost:8000/health` -- RabbitMQ: `amqp://localhost:5672` -- RabbitMQ management: `http://localhost:15672` +`JobDri`는 채용 공고부터 모의 지원, 자소서 분석까지 이어지는 지원 준비 흐름을 하나의 서비스 안에서 제공하는 백엔드입니다. -배포 서버 실행: - -```bash -cp .env.production.example .env -docker compose -f docker-compose.prod.yml up -d -``` +이 서버는 다음 흐름을 중심으로 동작합니다. -`prod` 프로필은 `/actuator/health`를 노출합니다. +1. **채용 공고 입력/정리** + 사용자가 텍스트 또는 이미지 기반 공고를 입력하면 AI가 회사명, 직무명, 주요 업무, 자격 요건, 우대 사항을 구조화합니다. +2. **공고 저장 및 모의 지원 생성** + 저장된 실제 공고 또는 AI가 만든 모의 공고를 기반으로 `MockApply`를 생성하고, 추천 문항 또는 직접 추가 문항을 선택해 지원서를 작성합니다. +3. **자소서 분석** + 작성한 답변을 AI가 비동기로 분석하고, 점수/피드백/문항별 코멘트/누락 키워드를 저장합니다. +4. **실시간 상태 추적** + 긴 작업은 `RabbitMQ + 외부 FastAPI worker + internal callback API + SSE` 구조로 처리하며, 사용자는 진행 상태와 완료 알림을 실시간으로 확인할 수 있습니다. -## Async Worker Architecture +## 🔄 Core Flow + +### 1. 채용 공고 처리 -현재 비동기 구조는 아래처럼 분리됩니다. +- `POST /api/job-postings/extract` + 텍스트 또는 S3 이미지 object key를 받아 AI가 공고를 추출합니다. +- `POST /api/job-postings/ingest` + 공고 추출 → 분류 후보 탐색 → 공고 초안 생성 → 저장까지 비동기 작업으로 접수합니다. +- `GET /api/job-postings/ingest/async/{taskId}` + 비동기 상태를 조회합니다. +- `GET /api/job-postings/ingest/async/{taskId}/stream` + SSE로 상태 변화를 구독합니다. -- API 서버: task row 생성, 상태 조회 API 제공, RabbitMQ publish, 최종 DB 저장 -- RabbitMQ: `job posting ingest` 작업 버퍼링과 재전달 -- FastAPI worker: MQ consume, OpenAI 기반 추출/분류/생성 수행, 내부 API callback +### 2. 모의 지원서 생성 + +- 저장된 실제 공고 기반 `ACTUAL` 타입 모의 지원 생성 +- 저장된 공고 기반 `MOCK` 타입 모의 지원 생성 +- 회사/직무 분류 기반 가상 공고 생성과 추천 질문 생성 +- 같은 공고 기준 회차(`sequence`) 관리 및 재도전 생성 -로컬에 필요한 추가 env: +### 3. 자소서 작성 및 분석 -```bash -RABBITMQ_HOST=localhost -RABBITMQ_PORT=5672 -RABBITMQ_USERNAME=guest -RABBITMQ_PASSWORD=guest -RABBITMQ_VHOST=/ -APP_WORKER_INTERNAL_API_KEY=change-me-internal-worker-key -APP_WORKER_JOB_POSTING_EXCHANGE=jobdri.worker.exchange -APP_WORKER_JOB_POSTING_QUEUE=jobdri.job-posting.ingest -APP_WORKER_JOB_POSTING_ROUTING_KEY=job-posting.ingest -APP_WORKER_JOB_POSTING_DLQ=jobdri.job-posting.ingest.dlq -WORKER_MAX_RETRY_COUNT=3 +- 문항 후보 조회, 직접 문항 추가, 선택 문항 저장 +- 답변 저장/수정 +- `POST /api/mock-applies/{mockApplyId}/analysis` + 자소서 분석 비동기 작업 접수 +- `GET /api/mock-applies/{mockApplyId}/analysis` + 저장된 분석 결과 조회 +- `GET /api/mock-applies/{mockApplyId}/analysis/async/{taskId}/stream` + SSE 상태 스트림 구독 + +## 🛠️ Tech Stack + +| Category | Stack | +| --- | --- | +| **Language** | Java 21 | +| **Framework** | Spring Boot 3.5.14 | +| **ORM / DB** | Spring Data JPA, PostgreSQL, H2(Test), pgvector | +| **Cache / Messaging** | Redis, RabbitMQ | +| **Auth / Security** | Spring Security, JWT, Google OAuth2 | +| **AI / Embedding** | OpenAI API (`gpt-4o-mini` default), Cohere Embedding (`embed-v4.0`) | +| **Storage / Infra** | AWS S3, Docker, GHCR, EC2, Nginx | +| **Payment** | Toss Payments | +| **Mail / Realtime** | JavaMailSender(Gmail SMTP), SSE | +| **Docs / Build** | SpringDoc Swagger, Gradle, GitHub Actions | + +## ✨ Key Features + +### 1. 인증 및 사용자 관리 + +- 이메일 인증번호 발송/검증 기반 회원가입 +- JWT Access/Refresh Token 로그인 및 재발급 +- Redis 기반 로그아웃/토큰 무효화 처리 +- Google OAuth2 로그인 및 프론트 리다이렉트 연동 + +### 2. 채용 공고 AI 처리 + +- 공고 텍스트/이미지에서 구조화 정보 추출 +- S3 presigned PUT URL 기반 이미지 직접 업로드 +- 직무 소분류 자동 분류 후보 추천 +- 회사/직무 기반 실제 공고 초안 생성 +- 회사/직무 기반 모의 공고 및 추천 질문 생성 + +### 3. 모의 지원서 작성 흐름 + +- 실제 공고 기반 `ACTUAL` 타입 지원 생성 +- 생성 공고 기반 `MOCK` 타입 지원 생성 +- 질문 후보 조회 및 직접 문항 추가 +- 선택 문항 저장 후 답변 작성 단계 전환 +- 공고별 회차 관리 및 재도전 지원 + +### 4. 자소서 분석 + +- 비동기 분석 작업 접수 및 상태 조회 +- 분석 완료 시 종합 점수, 항목별 점수, 전체 피드백 저장 +- 문항별 분석 결과와 누락 키워드 제공 +- 분석 실행 시 크레딧 1회 차감, 실패 시 환불 + +### 5. 결제 및 크레딧 + +- 크레딧 플랜 조회 +- 토스페이먼츠 결제 준비/승인 +- 사용자 크레딧 잔액/거래 내역 조회 +- 기본 플랜 + `1회권`, `5회권`, `10회권` + +### 6. 실시간 알림 + +- SSE 기반 인앱 알림 스트림 +- 미읽음 개수 조회 +- 개별/전체 읽음 처리 +- 현재 구현된 주요 알림 타입 + `JOB_POSTING_ASYNC_SUCCEEDED`, `JOB_POSTING_ASYNC_FAILED`, `ANALYSIS_ASYNC_SUCCEEDED`, `ANALYSIS_ASYNC_FAILED` + +### 7. 관리자 / 운영 기능 + +- corpus xlsx 적재 API +- corpus embedding 동기화 API +- 분석 retrieval preview 디버그 API +- 부트스트랩 관리자 이메일 승격 +- 비동기 task timeout sweep 스케줄러 +- 감사 로그 AOP (`@AuditLogEvent`) + +## 🔍 상세 기능 명세 + +### 🧩 인증 및 유저 + +- `POST /api/auth/email-verifications` +- `POST /api/auth/email-verifications/confirmations` +- `POST /api/auth/signup` +- `POST /api/auth/login` +- `POST /api/auth/reissue` +- `POST /api/auth/logout` +- `GET /oauth2/authorization/google` +- `GET /login/oauth2/code/google` + +### 🗂️ 직무 분류 + +- `GET /api/classifications` +- `GET /api/classifications/{bigId}/middles` +- `GET /api/classifications/middles/{middleId}/details` + +### 📝 채용 공고 + +- `POST /api/job-postings` +- `PATCH /api/job-postings/{jobPostingId}` +- `GET /api/job-postings/me` +- `GET /api/job-postings/me/{jobPostingId}` +- `DELETE /api/job-postings/{jobPostingId}` +- `POST /api/job-postings/generate` +- `POST /api/job-postings/extract` +- `POST /api/job-postings/ingest` +- `GET /api/job-postings/ingest/async/{taskId}` +- `GET /api/job-postings/ingest/async/{taskId}/stream` +- `POST /api/job-postings/images/presign-upload` +- `POST /api/job-postings/extension/ingest` + +### 🎯 모의 공고 / 모의 지원 + +- `POST /api/job-postings/mock/generate` +- `GET /api/job-postings/mock/questions` +- `POST /api/job-postings/mock/questions` +- `GET /api/mock-applies/me` +- `POST /api/mock-applies/actual` +- `POST /api/mock-applies/mock/from-job-posting` +- `POST /api/mock-applies/mock` +- `POST /api/mock-applies/{mockApplyId}/retry` +- `GET /api/mock-applies/{mockApplyId}/job-posting` +- `GET /api/mock-applies/{mockApplyId}/sequence` + +### ✍️ 문항 / 답변 / 분석 + +- `GET /api/mock-applies/{mockApplyId}/questions/candidates` +- `POST /api/mock-applies/{mockApplyId}/questions/candidates` +- `GET /api/mock-applies/{mockApplyId}/questions` +- `PUT /api/mock-applies/{mockApplyId}/questions` +- `PATCH /api/mock-applies/{mockApplyId}/questions/answers` +- `POST /api/mock-applies/{mockApplyId}/analysis` +- `GET /api/mock-applies/{mockApplyId}/analysis` +- `GET /api/mock-applies/{mockApplyId}/analysis/async/{taskId}` +- `GET /api/mock-applies/{mockApplyId}/analysis/async/{taskId}/stream` +- `GET /api/job-postings/{jobPostingId}/analysis` + +### 💳 결제 / 크레딧 + +- `GET /api/payments/plans` +- `POST /api/payments/prepare` +- `POST /api/payments/confirm` +- `GET /api/payments/credits/me` +- `GET /api/payments/credits/me/transactions` + +### 🔔 알림 + +- `GET /api/notifications` +- `GET /api/notifications/unread-count` +- `GET /api/notifications/stream` +- `PATCH /api/notifications/{notificationId}/read` +- `PATCH /api/notifications/read-all` + +### 🛠️ 관리자 / 워커 내부 API + +- `POST /api/admin/corpus/import` +- `POST /api/admin/corpus/import/upload` +- `POST /api/admin/corpus/embeddings/sync` +- `POST /api/admin/analysis/retrieval-preview` +- `POST /api/internal/worker/job-postings/...` +- `POST /api/internal/worker/analysis/...` + +## 🏛️ System Architecture + +### 1. 전체 서버 아키텍처 + +```mermaid +graph TD + Client[Web Client] -->|JWT / OAuth2 / REST| Spring[Spring Boot API] + Client -->|SSE| Spring + Client -->|Presigned URL Upload| S3[(AWS S3)] + + Spring -->|JPA| Postgres[(PostgreSQL)] + Spring -->|Cache / Token / Realtime state| Redis[(Redis)] + Spring -->|AMQP Publish| Rabbit[(RabbitMQ)] + Spring -->|OpenAI Responses API| OpenAI[OpenAI] + Spring -->|Embedding Sync| Cohere[Cohere] + Spring -->|Payment Confirm| Toss[Toss Payments] + Spring -->|SMTP| Gmail[Gmail SMTP] + + Rabbit -->|Task Consume| Worker[FastAPI Worker
external deployment] + Worker -->|Internal callback API| Spring + + GitHub[GitHub Actions] -->|Build jar / Docker image| GHCR[GHCR] + GHCR -->|Pull latest image| EC2[EC2 + Docker Compose] + EC2 -->|Reverse Proxy| Nginx[Nginx] + Nginx --> Spring ``` -스모크 테스트: +### 2. 비동기 워커 처리 흐름 + +```mermaid +sequenceDiagram + participant U as User + participant C as Client + participant S as Spring Boot API + participant MQ as RabbitMQ + participant W as FastAPI Worker + participant DB as PostgreSQL + participant SSE as SSE Stream + + U->>C: 채용 공고 분석 / 자소서 분석 요청 + C->>S: 비동기 API 호출 + S->>DB: async task 생성 + S->>MQ: 작업 메시지 publish + S-->>C: taskId 반환 + + C->>SSE: taskId 기준 상태 스트림 구독 + MQ->>W: 작업 전달 + W->>S: context 조회 (internal API) + W->>S: running / retry / failed / complete callback + S->>DB: 상태 및 결과 저장 + S-->>SSE: 상태 변경 이벤트 전송 + SSE-->>C: 진행 상태 / 완료 / 실패 수신 +``` -1. `docker compose up --build` -2. 로그인 토큰을 준비합니다. -3. `POST /api/job-postings/ingest` 로 `rawText` 또는 `imageObjectKey`를 전송합니다. -4. 응답의 `taskId`로 `GET /api/job-postings/ingest/async/{taskId}` 를 조회합니다. -5. worker가 성공하면 `SUCCEEDED`, 실패하면 `FAILED` 와 에러 메시지가 반환됩니다. +## 🔁 Async Worker Flow -worker 단독 실행: +이 서버의 핵심 비동기 작업은 `채용 공고 ingest`와 `자소서 분석` 두 종류입니다. -```bash -python3 -m venv .venv -source .venv/bin/activate -pip install -r worker/requirements.txt -uvicorn worker.app.main:app --host 0.0.0.0 --port 8000 -``` +1. 사용자가 비동기 API를 호출합니다. +2. Spring 서버가 task row를 생성합니다. +3. Spring 서버가 RabbitMQ exchange/queue로 작업 메시지를 publish 합니다. +4. 외부 FastAPI worker가 메시지를 consume 합니다. +5. worker는 필요 컨텍스트를 internal API로 조회합니다. +6. worker가 AI 처리 결과를 internal callback API로 완료/실패 반영합니다. +7. Spring 서버가 DB 저장, 알림 생성, SSE 상태 전파를 수행합니다. -## Corpus Import Script +> 현재 이 저장소에는 **Spring Boot API 서버 코드만 포함**되어 있습니다. +> `RabbitMQ`로 연동되는 **FastAPI worker는 외부 배포 단위**로 전제되어 있으며, +> `docker-compose.yml`에는 worker 서비스가 정의되어 있지만 실제 `worker/` 디렉터리는 이 리포지토리에 포함되어 있지 않습니다. -대용량 corpus 엑셀 적재는 관리자 API 대신 Python 스크립트로 직접 실행할 수 있습니다. - -설치: +## 📂 Project Structure ```bash -python3 -m venv .venv -source .venv/bin/activate -pip install -r scripts/requirements-corpus-import.txt +com.jobdri.jobdri_api +├── domain +│ ├── analysis # 자소서 분석, 질문 선택, async task, worker bridge +│ ├── auth # 회원가입, 로그인, 이메일 인증, OAuth2 +│ ├── classification # 대/중/소분류 직무 체계 조회 +│ ├── company # 회사 엔티티 +│ ├── corpus # corpus 적재, 임베딩 동기화, retrieval +│ ├── jobposting # 공고 저장/수정/생성/추출/ingest/extension 연동 +│ ├── mockapply # 모의 지원 생성, 회차, 재도전 +│ ├── notification # SSE 알림, 읽음 처리 +│ ├── payment # 크레딧, 결제, 거래 내역 +│ ├── audit # 감사 로그 AOP +│ ├── user # 사용자 도메인 +│ └── skill / experience +├── global +│ ├── apiPayload # 공통 응답/에러 코드 +│ ├── config # Security, Swagger, RabbitMQ, AWS S3, Async 설정 +│ ├── jwt # JWT 필터/유틸 +│ ├── mq # RabbitMQ publish support +│ ├── scheduling # async timeout sweep scheduler +│ ├── security # UserDetails, internal API key validator +│ └── sse # SSE subscription registry +└── resources + ├── application*.yaml + └── schema.sql ``` -실행: +## 🚀 Local Run + +### 1. 애플리케이션 실행 ```bash cp .env.example .env -python scripts/import_corpus.py /absolute/path/to/file.xlsx +./gradlew bootRun ``` -- `.env`의 `DB_URL`, `DB_USERNAME`, `DB_PASSWORD`를 사용합니다. -- 기본 env 파일 경로를 바꾸려면 `--env-file /path/to/.env` 옵션을 사용할 수 있습니다. -- 엑셀 시트명은 `jd_embed_corpus`, `question_embed_corpus`를 사용합니다. -- `source_analysis_id`, `source_question_id` 기준으로 `INSERT ... ON CONFLICT DO UPDATE` 방식으로 적재합니다. - -## Corpus Embedding Sync Script +기본 포트: -관리자 API 대신 Python 스크립트로 corpus 임베딩을 일괄 동기화할 수 있습니다. +- API: `http://localhost:8080` +- Swagger: `http://localhost:8080/swagger-ui/index.html` -실행: +### 2. 테스트 실행 ```bash -source .venv/bin/activate -pip install -r scripts/requirements-corpus-import.txt -python scripts/sync_corpus_embeddings.py --env-file .env +./gradlew test ``` -옵션 예시: +테스트 환경은 `H2 (MODE=PostgreSQL)`를 사용합니다. + +### 3. Docker Compose ```bash -python scripts/sync_corpus_embeddings.py --env-file .env --limit 100 -python scripts/sync_corpus_embeddings.py --env-file .env --job-only -python scripts/sync_corpus_embeddings.py --env-file .env --question-only --batch-size 16 +cp .env.example .env +docker compose up --build ``` -- `.env`의 `DB_URL`, `DB_USERNAME`, `DB_PASSWORD`, `COHERE_API_KEY`를 사용합니다. -- 기본 모델은 `embed-v4.0`, 기본 배치 크기는 `32`입니다. -- `mock_job_posting_embeddings`, `mock_question_embeddings` 테이블에 `INSERT ... ON CONFLICT DO UPDATE` 방식으로 적재합니다. +포함 의도 서비스: -## CI/CD +- Spring Boot API +- PostgreSQL +- Redis +- RabbitMQ +- external FastAPI worker 연결용 worker service 정의 -- `CI`: `main`, `develop` 브랜치 push 및 PR에서 테스트와 Docker 이미지 빌드를 실행합니다. -- `Deploy`: `main` 브랜치 push 또는 수동 실행 시 GHCR에 이미지를 푸시하고, 배포 서버 secret이 있으면 SSH로 `docker-compose.prod.yml`을 갱신합니다. +주의사항: -GitHub Actions 배포 secret: +- 현재 `docker-compose.yml`은 `worker/Dockerfile`을 참조하지만, 이 리포지토리에는 `worker/` 디렉터리가 없습니다. +- 따라서 로컬에서 compose 전체를 그대로 올리려면 worker 구현체 또는 별도 이미지 준비가 필요합니다. -- `DEPLOY_HOST` -- `DEPLOY_USER` -- `DEPLOY_PORT` optional, default `22` -- `DEPLOY_SSH_KEY` -- `DEPLOY_PATH` -- `GHCR_USERNAME` -- `GHCR_TOKEN` +## ⚙️ Environment -## 배포 환경 +주요 환경 변수: -백엔드 API는 Docker 이미지로 빌드되어 GHCR에 업로드되고, GitHub Actions를 통해 EC2 서버에 배포됩니다. +- `DB_URL`, `DB_USERNAME`, `DB_PASSWORD` +- `REDIS_HOST`, `REDIS_PORT` +- `RABBITMQ_HOST`, `RABBITMQ_PORT`, `RABBITMQ_USERNAME`, `RABBITMQ_PASSWORD` +- `JWT_SECRET_KEY`, `JWT_ACCESS_TOKEN_EXPIRATION`, `JWT_REFRESH_TOKEN_EXPIRATION` +- `MAIL_HOST`, `MAIL_USERNAME`, `MAIL_PASSWORD` +- `GOOGLE_CLIENT_ID`, `GOOGLE_CLIENT_SECRET` +- `OPENAI_API_KEY` +- `COHERE_API_KEY` +- `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_REGION`, `S3_BUCKET` +- `TOSS_CLIENT_KEY`, `TOSS_SECRET_KEY` +- `APP_WORKER_INTERNAL_API_KEY` -### 인프라 +worker 연동 관련 기본 queue 설정: -- 서버: AWS EC2 `t3.small` -- OS: Amazon Linux 2023 -- DB: AWS RDS PostgreSQL `db.t3.micro` -- Redis: Upstash Redis -- 웹 서버: Nginx -- 컨테이너 런타임: Docker, Docker Compose +- `APP_WORKER_JOB_POSTING_QUEUE=jobdri.job-posting.ingest` +- `APP_WORKER_ANALYSIS_QUEUE=jobdri.analysis.execute` -### 배포 흐름 +## 🧠 Corpus / Retrieval -1. `main` 브랜치에 변경사항이 반영됩니다. -2. GitHub Actions가 Docker 이미지를 빌드합니다. -3. 이미지를 GHCR에 푸시합니다. -4. GitHub Actions가 EC2에 SSH로 접속합니다. -5. EC2에서 `docker-compose.prod.yml`을 사용해 최신 이미지를 pull하고 API 컨테이너를 재기동합니다. +이 프로젝트는 단순 LLM 호출만 사용하는 구조가 아니라, corpus 기반 retrieval를 함께 사용합니다. -### 서버 배포 경로 +- 채용 공고 corpus 적재 +- 질문 corpus 적재 +- Cohere 임베딩 생성 +- pgvector 저장 +- 모의 공고 생성 및 자소서 분석 시 retrieval context 참고 +- 관리자 preview API로 retrieval 결과 사전 점검 가능 -```bash -/opt/jobdri-api -``` +Python 스크립트도 포함되어 있습니다. -서버에는 아래 파일이 필요합니다. +- `scripts/import_corpus.py` +- `scripts/sync_corpus_embeddings.py` -```text -docker-compose.prod.yml -.env -``` +## 🚀 Deployment Pipeline -## 운영 API 도메인 +GitHub Actions 기준 배포 흐름: -운영 API는 Nginx reverse proxy와 Let's Encrypt 인증서를 통해 HTTPS로 제공됩니다. +1. `main`, `dev` 브랜치 push 또는 PR 발생 +2. CI에서 JDK 21 기반 `./gradlew clean test` 실행 +3. Docker image build +4. Deploy workflow에서 GHCR에 이미지 push +5. 배포 secret이 존재하면 원격 서버에 SSH 접속 +6. 서버에서 `docker compose -f docker-compose.prod.yml pull api` +7. `docker compose -f docker-compose.prod.yml up -d api` -```text -https://api.jobdri.site -``` +현재 workflow 기준 특징: -- Nginx는 80/443 요청을 Spring Boot API 컨테이너의 8080 포트로 프록시합니다. -- SSL 인증서는 Certbot으로 발급했으며 자동 갱신이 설정되어 있습니다. -- 상태 확인: `https://api.jobdri.site/actuator/health` +- 배포 이미지는 `ghcr.io/jobdri-developer/backend` +- `docker-compose.prod.yml`은 **API 컨테이너 단독 배포** 기준 +- worker 배포는 이 리포지토리 workflow에 포함되어 있지 않습니다 -정상 응답: +## 📝 API Documentation -```json -{"status":"UP","groups":["liveness","readiness"]} -``` +SpringDoc Swagger UI: + +- Local: `http://localhost:8080/swagger-ui/index.html` +- Production: `https://api.jobdri.site/swagger-ui/index.html` + +추가 공개 경로: + +- OpenAPI docs: `/v3/api-docs` +- Health check: `/actuator/health` + +## 📌 현재 구현 범위 + +- 이메일 인증 기반 로컬 회원가입/로그인 +- Google OAuth2 로그인 +- JWT 인증/인가 및 로그아웃 처리 +- 직무 대/중/소분류 조회 +- 채용 공고 생성/수정/조회/삭제 +- 채용 공고 이미지 presigned upload +- AI 채용 공고 추출 +- AI 채용 공고 초안 생성 +- AI 모의 공고 생성 +- 추천 질문 캐시 및 조회 +- 실제/모의 공고 기반 모의 지원 생성 +- 질문 선택, 직접 문항 추가, 답변 저장 +- 자소서 분석 비동기 처리 +- 크레딧 차감/환불 +- 토스 결제 연동 +- SSE 알림 +- 관리자 corpus import / embedding sync +- RabbitMQ 기반 외부 worker 연동용 internal API +- async timeout sweep scheduler +- 감사 로그 적재 diff --git a/ops/db/migrations/20260707_notifications.sql b/ops/db/migrations/20260707_notifications.sql new file mode 100644 index 0000000..b290381 --- /dev/null +++ b/ops/db/migrations/20260707_notifications.sql @@ -0,0 +1,21 @@ +CREATE TABLE IF NOT EXISTS notifications ( + id BIGSERIAL PRIMARY KEY, + user_id BIGINT NOT NULL, + notification_type VARCHAR(50) NOT NULL, + title VARCHAR(120) NOT NULL, + body VARCHAR(500) NOT NULL, + target_type VARCHAR(50) NOT NULL, + target_id VARCHAR(100), + payload_json TEXT, + read_at TIMESTAMP, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP, + CONSTRAINT fk_notifications_user FOREIGN KEY (user_id) REFERENCES users (id) +); + +CREATE INDEX IF NOT EXISTS idx_notifications_user_created_at + ON notifications (user_id, created_at DESC); + +CREATE INDEX IF NOT EXISTS idx_notifications_user_read_at + ON notifications (user_id, read_at) + WHERE read_at IS NULL; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java index c9970d1..5eb3f75 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java @@ -5,10 +5,14 @@ import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.CreditStatus; import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.FailureReason; import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.TaskStatus; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationTargetType; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationType; +import com.jobdri.jobdri_api.domain.notification.service.NotificationService; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode; import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -18,14 +22,18 @@ import java.time.Instant; import java.util.EnumSet; import java.util.Optional; +import java.util.LinkedHashMap; +import java.util.Map; @Service @RequiredArgsConstructor +@Slf4j // 분석 비동기 task 엔티티의 생성, 상태 전이, 조회를 전담하는 서비스다. public class AnalysisAsyncTaskService { private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; private final AnalysisAsyncSseService analysisAsyncSseService; + private final NotificationService notificationService; @Value("${app.worker.analysis.max-retry-count:3}") private int maxRetryCount; @@ -63,6 +71,7 @@ public void markSuccess(String taskId) { AnalysisAsyncTask task = getTask(taskId); task.markSuccess(); publishAfterCommit(toStatusResponse(task)); + createSuccessNotificationSafely(task); } @Transactional @@ -77,6 +86,7 @@ public void markFailed(String taskId, FailureReason failureReason, String errorM AnalysisAsyncTask task = getTask(taskId); task.markFailed(failureReason, errorMessage, retryCount); publishAfterCommit(toStatusResponse(task)); + createFailureNotificationSafely(task); } @Transactional @@ -160,4 +170,62 @@ public void afterCommit() { } }); } + + private void createSuccessNotificationSafely(AnalysisAsyncTask task) { + try { + createSuccessNotification(task); + } catch (Exception e) { + log.warn("분석 완료 알림 생성에 실패했습니다. taskId={}, userId={}", task.getTaskId(), task.getUserId(), e); + } + } + + private void createFailureNotificationSafely(AnalysisAsyncTask task) { + try { + createFailureNotification(task); + } catch (Exception e) { + log.warn( + "분석 실패 알림 생성에 실패했습니다. taskId={}, userId={}, error={}", + task.getTaskId(), + task.getUserId(), + task.getError(), + e + ); + } + } + + private void createSuccessNotification(AnalysisAsyncTask task) { + Map payload = new LinkedHashMap<>(); + payload.put("taskId", task.getTaskId()); + payload.put("mockApplyId", task.getMockApplyId()); + payload.put("status", task.getStatus().name()); + + notificationService.createNotification( + task.getUserId(), + NotificationType.ANALYSIS_ASYNC_SUCCEEDED, + "자소서 분석이 완료되었습니다.", + "분석 결과를 확인해보세요.", + NotificationTargetType.ANALYSIS_RESULT, + String.valueOf(task.getMockApplyId()), + payload + ); + } + + private void createFailureNotification(AnalysisAsyncTask task) { + String userFacingMessage = "자소서 분석 처리 중 오류가 발생했습니다."; + Map payload = new LinkedHashMap<>(); + payload.put("taskId", task.getTaskId()); + payload.put("mockApplyId", task.getMockApplyId()); + payload.put("failureReason", task.getFailureReason() != null ? task.getFailureReason().name() : null); + payload.put("status", task.getStatus().name()); + + notificationService.createNotification( + task.getUserId(), + NotificationType.ANALYSIS_ASYNC_FAILED, + "자소서 분석이 실패했습니다.", + userFacingMessage, + NotificationTargetType.ANALYSIS_TASK, + task.getTaskId(), + payload + ); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java index 266b8fc..52275a5 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java @@ -5,6 +5,9 @@ import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask; import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.FailureReason; import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.TaskStatus; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationTargetType; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationType; +import com.jobdri.jobdri_api.domain.notification.service.NotificationService; import com.jobdri.jobdri_api.domain.jobposting.repository.JobPostingAsyncTaskRepository; import com.jobdri.jobdri_api.domain.user.entity.User; import com.jobdri.jobdri_api.domain.user.entity.UserRole; @@ -13,6 +16,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -23,14 +27,18 @@ import java.time.Instant; import java.time.LocalDateTime; import java.util.EnumSet; +import java.util.LinkedHashMap; +import java.util.Map; @Service @RequiredArgsConstructor +@Slf4j public class JobPostingAsyncTaskService { private final JobPostingAsyncTaskRepository jobPostingAsyncTaskRepository; private final ObjectMapper objectMapper; private final JobPostingAsyncSseService jobPostingAsyncSseService; + private final NotificationService notificationService; @Value("${app.worker.job-posting.max-retry-count:3}") private int maxRetryCount; @@ -75,6 +83,7 @@ public JobPostingIngestResponse markSuccess(String taskId, JobPostingIngestRespo } task.markSuccess(serializeResult(result)); publishAfterCommit(toStatusResponse(task)); + createSuccessNotificationSafely(task, result); return result; } @@ -96,6 +105,7 @@ public void markFailed(String taskId, FailureReason failureReason, String errorM } task.markFailed(failureReason, errorMessage, retryCount); publishAfterCommit(toStatusResponse(task)); + createFailureNotificationSafely(task); } @Transactional @@ -242,4 +252,68 @@ public void afterCommit() { } }); } + + private void createSuccessNotificationSafely(JobPostingAsyncTask task, JobPostingIngestResponse result) { + try { + createSuccessNotification(task, result); + } catch (Exception e) { + log.warn("채용 공고 완료 알림 생성에 실패했습니다. taskId={}, userId={}", task.getTaskId(), task.getUserId(), e); + } + } + + private void createFailureNotificationSafely(JobPostingAsyncTask task) { + try { + createFailureNotification(task); + } catch (Exception e) { + log.warn( + "채용 공고 실패 알림 생성에 실패했습니다. taskId={}, userId={}, error={}", + task.getTaskId(), + task.getUserId(), + task.getError(), + e + ); + } + } + + private void createSuccessNotification(JobPostingAsyncTask task, JobPostingIngestResponse result) { + boolean hasSavedJobPosting = result.getSaved() != null && result.getSaved().getJobPostingId() != null; + Map payload = new LinkedHashMap<>(); + payload.put("taskId", task.getTaskId()); + payload.put("savedToDatabase", result.isSavedToDatabase()); + payload.put("jobPostingId", result.getSaved() != null ? result.getSaved().getJobPostingId() : null); + + notificationService.createNotification( + task.getUserId(), + NotificationType.JOB_POSTING_ASYNC_SUCCEEDED, + "채용 공고 작업이 완료되었습니다.", + result.isSavedToDatabase() + ? "채용 공고 분석과 저장이 완료되었습니다." + : "채용 공고 분석이 완료되었습니다.", + hasSavedJobPosting + ? NotificationTargetType.JOB_POSTING_RESULT + : NotificationTargetType.JOB_POSTING_TASK, + hasSavedJobPosting + ? String.valueOf(result.getSaved().getJobPostingId()) + : task.getTaskId(), + payload + ); + } + + private void createFailureNotification(JobPostingAsyncTask task) { + String userFacingMessage = "채용 공고 작업 처리 중 오류가 발생했습니다."; + Map payload = new LinkedHashMap<>(); + payload.put("taskId", task.getTaskId()); + payload.put("failureReason", task.getFailureReason() != null ? task.getFailureReason().name() : null); + payload.put("status", task.getStatus().name()); + + notificationService.createNotification( + task.getUserId(), + NotificationType.JOB_POSTING_ASYNC_FAILED, + "채용 공고 작업이 실패했습니다.", + userFacingMessage, + NotificationTargetType.JOB_POSTING_TASK, + task.getTaskId(), + payload + ); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/controller/NotificationController.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/controller/NotificationController.java new file mode 100644 index 0000000..fbe815d --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/controller/NotificationController.java @@ -0,0 +1,89 @@ +package com.jobdri.jobdri_api.domain.notification.controller; + +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationReadAllResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationUnreadCountResponse; +import com.jobdri.jobdri_api.domain.notification.service.NotificationService; +import com.jobdri.jobdri_api.global.apiPayload.ApiResponse; +import com.jobdri.jobdri_api.global.security.UserDetailsImpl; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import lombok.RequiredArgsConstructor; +import org.springframework.http.CacheControl; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PatchMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.List; + +@RestController +@RequiredArgsConstructor +@RequestMapping("/api/notifications") +@Tag(name = "Notification", description = "인앱 알림 API") +public class NotificationController { + + private final NotificationService notificationService; + + @Operation(summary = "알림 목록 조회", description = "최근 알림 목록을 조회합니다.") + @GetMapping + public ApiResponse> getNotifications( + @AuthenticationPrincipal UserDetailsImpl userDetails + ) { + return ApiResponse.onSuccess( + "알림 목록 조회에 성공했습니다.", + notificationService.getNotifications(userDetails.getUser()) + ); + } + + @Operation(summary = "미읽음 알림 개수 조회", description = "현재 미읽음 알림 개수를 조회합니다.") + @GetMapping("/unread-count") + public ApiResponse getUnreadCount( + @AuthenticationPrincipal UserDetailsImpl userDetails + ) { + return ApiResponse.onSuccess( + "미읽음 알림 개수 조회에 성공했습니다.", + notificationService.getUnreadCount(userDetails.getUser()) + ); + } + + @Operation(summary = "알림 SSE 구독", description = "실시간 알림 스트림을 구독합니다.") + @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity streamNotifications( + @AuthenticationPrincipal UserDetailsImpl userDetails + ) { + return ResponseEntity.ok() + .header(HttpHeaders.CACHE_CONTROL, CacheControl.noCache().getHeaderValue()) + .header("X-Accel-Buffering", "no") + .body(notificationService.subscribe(userDetails.getUser())); + } + + @Operation(summary = "알림 읽음 처리", description = "특정 알림을 읽음 처리합니다.") + @PatchMapping("/{notificationId}/read") + public ApiResponse markRead( + @AuthenticationPrincipal UserDetailsImpl userDetails, + @PathVariable Long notificationId + ) { + return ApiResponse.onSuccess( + "알림 읽음 처리에 성공했습니다.", + notificationService.markRead(userDetails.getUser(), notificationId) + ); + } + + @Operation(summary = "알림 전체 읽음 처리", description = "미읽음 알림을 모두 읽음 처리합니다.") + @PatchMapping("/read-all") + public ApiResponse markAllRead( + @AuthenticationPrincipal UserDetailsImpl userDetails + ) { + return ApiResponse.onSuccess( + "알림 전체 읽음 처리에 성공했습니다.", + notificationService.markAllRead(userDetails.getUser()) + ); + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationReadAllResponse.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationReadAllResponse.java new file mode 100644 index 0000000..f4e15b3 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationReadAllResponse.java @@ -0,0 +1,4 @@ +package com.jobdri.jobdri_api.domain.notification.dto.response; + +public record NotificationReadAllResponse(int updatedCount, long unreadCount) { +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationResponse.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationResponse.java new file mode 100644 index 0000000..6951a50 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationResponse.java @@ -0,0 +1,23 @@ +package com.jobdri.jobdri_api.domain.notification.dto.response; + +import com.jobdri.jobdri_api.domain.notification.entity.NotificationTargetType; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationType; +import lombok.Builder; + +import java.time.LocalDateTime; +import java.util.Map; + +@Builder +public record NotificationResponse( + Long id, + NotificationType type, + String title, + String body, + boolean isRead, + LocalDateTime readAt, + LocalDateTime createdAt, + NotificationTargetType targetType, + String targetId, + Map payload +) { +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationStreamBootstrapResponse.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationStreamBootstrapResponse.java new file mode 100644 index 0000000..84644f8 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationStreamBootstrapResponse.java @@ -0,0 +1,9 @@ +package com.jobdri.jobdri_api.domain.notification.dto.response; + +import java.util.List; + +public record NotificationStreamBootstrapResponse( + long unreadCount, + List notifications +) { +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationStreamEventResponse.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationStreamEventResponse.java new file mode 100644 index 0000000..9e5ae86 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationStreamEventResponse.java @@ -0,0 +1,11 @@ +package com.jobdri.jobdri_api.domain.notification.dto.response; + +import java.util.List; + +public record NotificationStreamEventResponse( + String action, + NotificationResponse notification, + List notificationIds, + long unreadCount +) { +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationUnreadCountResponse.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationUnreadCountResponse.java new file mode 100644 index 0000000..ffbb240 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/dto/response/NotificationUnreadCountResponse.java @@ -0,0 +1,4 @@ +package com.jobdri.jobdri_api.domain.notification.dto.response; + +public record NotificationUnreadCountResponse(long unreadCount) { +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/Notification.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/Notification.java new file mode 100644 index 0000000..7f6fa62 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/Notification.java @@ -0,0 +1,88 @@ +package com.jobdri.jobdri_api.domain.notification.entity; + +import com.jobdri.jobdri_api.domain.user.entity.User; +import com.jobdri.jobdri_api.global.entity.BaseEntity; +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.FetchType; +import jakarta.persistence.GeneratedValue; +import jakarta.persistence.GenerationType; +import jakarta.persistence.Id; +import jakarta.persistence.JoinColumn; +import jakarta.persistence.ManyToOne; +import jakarta.persistence.Table; +import lombok.AccessLevel; +import lombok.Getter; +import lombok.NoArgsConstructor; + +import java.time.LocalDateTime; + +@Entity +@Getter +@NoArgsConstructor(access = AccessLevel.PROTECTED) +@Table(name = "notifications") +public class Notification extends BaseEntity { + + @Id + @GeneratedValue(strategy = GenerationType.IDENTITY) + private Long id; + + @ManyToOne(fetch = FetchType.LAZY, optional = false) + @JoinColumn(name = "user_id", nullable = false) + private User user; + + @Enumerated(EnumType.STRING) + @Column(name = "notification_type", nullable = false, length = 50) + private NotificationType notificationType; + + @Column(nullable = false, length = 120) + private String title; + + @Column(nullable = false, length = 500) + private String body; + + @Enumerated(EnumType.STRING) + @Column(name = "target_type", nullable = false, length = 50) + private NotificationTargetType targetType; + + @Column(name = "target_id", length = 100) + private String targetId; + + @Column(name = "payload_json", columnDefinition = "TEXT") + private String payloadJson; + + @Column(name = "read_at") + private LocalDateTime readAt; + + public static Notification create( + User user, + NotificationType notificationType, + String title, + String body, + NotificationTargetType targetType, + String targetId, + String payloadJson + ) { + Notification notification = new Notification(); + notification.user = user; + notification.notificationType = notificationType; + notification.title = title; + notification.body = body; + notification.targetType = targetType; + notification.targetId = targetId; + notification.payloadJson = payloadJson; + return notification; + } + + public boolean isRead() { + return readAt != null; + } + + public void markRead() { + if (readAt == null) { + readAt = LocalDateTime.now(); + } + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/NotificationTargetType.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/NotificationTargetType.java new file mode 100644 index 0000000..1b15264 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/NotificationTargetType.java @@ -0,0 +1,9 @@ +package com.jobdri.jobdri_api.domain.notification.entity; + +public enum NotificationTargetType { + JOB_POSTING_TASK, + JOB_POSTING_RESULT, + ANALYSIS_TASK, + ANALYSIS_RESULT, + NONE +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/NotificationType.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/NotificationType.java new file mode 100644 index 0000000..e129d5d --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/entity/NotificationType.java @@ -0,0 +1,9 @@ +package com.jobdri.jobdri_api.domain.notification.entity; + +public enum NotificationType { + JOB_POSTING_ASYNC_SUCCEEDED, + JOB_POSTING_ASYNC_FAILED, + ANALYSIS_ASYNC_SUCCEEDED, + ANALYSIS_ASYNC_FAILED, + GENERAL +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/repository/NotificationRepository.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/repository/NotificationRepository.java new file mode 100644 index 0000000..1a11522 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/repository/NotificationRepository.java @@ -0,0 +1,28 @@ +package com.jobdri.jobdri_api.domain.notification.repository; + +import com.jobdri.jobdri_api.domain.notification.entity.Notification; +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Optional; + +public interface NotificationRepository extends JpaRepository { + List findTop50ByUserIdOrderByCreatedAtDesc(Long userId); + + Optional findByIdAndUserId(Long id, Long userId); + + long countByUserIdAndReadAtIsNull(Long userId); + + @Modifying(clearAutomatically = true, flushAutomatically = true) + @Query(""" + update Notification n + set n.readAt = :readAt + where n.user.id = :userId + and n.readAt is null + """) + int markAllAsRead(@Param("userId") Long userId, @Param("readAt") LocalDateTime readAt); +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/service/NotificationService.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/service/NotificationService.java new file mode 100644 index 0000000..3dc0f93 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/service/NotificationService.java @@ -0,0 +1,198 @@ +package com.jobdri.jobdri_api.domain.notification.service; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationReadAllResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationStreamBootstrapResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationStreamEventResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationUnreadCountResponse; +import com.jobdri.jobdri_api.domain.notification.entity.Notification; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationTargetType; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationType; +import com.jobdri.jobdri_api.domain.notification.repository.NotificationRepository; +import com.jobdri.jobdri_api.domain.user.entity.User; +import com.jobdri.jobdri_api.domain.user.service.UserService; +import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode; +import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Propagation; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + +@Service +@RequiredArgsConstructor +public class NotificationService { + + private static final TypeReference> PAYLOAD_TYPE = new TypeReference<>() { + }; + + private final NotificationRepository notificationRepository; + private final NotificationSseService notificationSseService; + private final UserService userService; + private final ObjectMapper objectMapper; + + @Transactional(readOnly = true) + public List getNotifications(User user) { + User validatedUser = userService.validateUser(user); + return notificationRepository.findTop50ByUserIdOrderByCreatedAtDesc(validatedUser.getId()).stream() + .map(this::toResponse) + .toList(); + } + + @Transactional(readOnly = true) + public NotificationUnreadCountResponse getUnreadCount(User user) { + User validatedUser = userService.validateUser(user); + return new NotificationUnreadCountResponse(notificationRepository.countByUserIdAndReadAtIsNull(validatedUser.getId())); + } + + @Transactional(readOnly = true) + public SseEmitter subscribe(User user) { + User validatedUser = userService.validateUser(user); + return notificationSseService.subscribe( + validatedUser.getId(), + () -> new NotificationStreamBootstrapResponse( + notificationRepository.countByUserIdAndReadAtIsNull(validatedUser.getId()), + notificationRepository.findTop50ByUserIdOrderByCreatedAtDesc(validatedUser.getId()).stream() + .map(this::toResponse) + .toList() + ) + ); + } + + @Transactional + public NotificationResponse markRead(User user, Long notificationId) { + User validatedUser = userService.validateUser(user); + Notification notification = notificationRepository.findByIdAndUserId(notificationId, validatedUser.getId()) + .orElseThrow(() -> new GeneralException( + GeneralErrorCode.INVALID_PARAMETER, + "해당 알림을 찾을 수 없습니다. notificationId=" + notificationId + )); + boolean wasUnread = !notification.isRead(); + notification.markRead(); + + NotificationResponse response = toResponse(notification); + if (wasUnread) { + publishAfterCommit( + validatedUser.getId(), + () -> new NotificationStreamEventResponse( + "READ", + response, + List.of(notification.getId()), + notificationRepository.countByUserIdAndReadAtIsNull(validatedUser.getId()) + ) + ); + } + return response; + } + + @Transactional + public NotificationReadAllResponse markAllRead(User user) { + User validatedUser = userService.validateUser(user); + int updatedCount = notificationRepository.markAllAsRead(validatedUser.getId(), LocalDateTime.now()); + if (updatedCount > 0) { + publishAfterCommit( + validatedUser.getId(), + () -> new NotificationStreamEventResponse( + "READ_ALL", + null, + null, + notificationRepository.countByUserIdAndReadAtIsNull(validatedUser.getId()) + ) + ); + } + return new NotificationReadAllResponse(updatedCount, notificationRepository.countByUserIdAndReadAtIsNull(validatedUser.getId())); + } + + @Transactional(propagation = Propagation.REQUIRES_NEW) + public NotificationResponse createNotification( + Long userId, + NotificationType notificationType, + String title, + String body, + NotificationTargetType targetType, + String targetId, + Map payload + ) { + User user = userService.getUser(userId); + Notification notification = notificationRepository.save( + Notification.create( + user, + notificationType, + title, + body, + targetType, + targetId, + serializePayload(payload) + ) + ); + + NotificationResponse response = toResponse(notification); + publishAfterCommit( + userId, + () -> new NotificationStreamEventResponse( + "CREATED", + response, + List.of(notification.getId()), + notificationRepository.countByUserIdAndReadAtIsNull(userId) + ) + ); + return response; + } + + private void publishAfterCommit(Long userId, Supplier eventSupplier) { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + notificationSseService.publish(userId, eventSupplier.get()); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + notificationSseService.publish(userId, eventSupplier.get()); + } + }); + } + + private NotificationResponse toResponse(Notification notification) { + return NotificationResponse.builder() + .id(notification.getId()) + .type(notification.getNotificationType()) + .title(notification.getTitle()) + .body(notification.getBody()) + .isRead(notification.isRead()) + .readAt(notification.getReadAt()) + .createdAt(notification.getCreatedAt()) + .targetType(notification.getTargetType()) + .targetId(notification.getTargetId()) + .payload(readPayload(notification.getPayloadJson())) + .build(); + } + + private String serializePayload(Map payload) { + try { + return payload == null || payload.isEmpty() ? null : objectMapper.writeValueAsString(payload); + } catch (JsonProcessingException e) { + throw new GeneralException(GeneralErrorCode.INTERNAL_SERVER_ERROR, "알림 payload 직렬화에 실패했습니다."); + } + } + + private Map readPayload(String payloadJson) { + if (payloadJson == null || payloadJson.isBlank()) { + return Map.of(); + } + try { + return objectMapper.readValue(payloadJson, PAYLOAD_TYPE); + } catch (JsonProcessingException e) { + return Map.of(); + } + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/notification/service/NotificationSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/notification/service/NotificationSseService.java new file mode 100644 index 0000000..79fb918 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/notification/service/NotificationSseService.java @@ -0,0 +1,37 @@ +package com.jobdri.jobdri_api.domain.notification.service; + +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationStreamBootstrapResponse; +import com.jobdri.jobdri_api.domain.notification.dto.response.NotificationStreamEventResponse; +import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.function.Supplier; + +@Service +@RequiredArgsConstructor +public class NotificationSseService { + + private static final String BOOTSTRAP_EVENT_NAME = "notification-bootstrap"; + private static final String NOTIFICATION_EVENT_NAME = "notification"; + + private final SseSubscriptionRegistry sseSubscriptionRegistry; + + public SseEmitter subscribe(Long userId, Supplier bootstrapSupplier) { + return sseSubscriptionRegistry.subscribe( + channelKey(userId), + BOOTSTRAP_EVENT_NAME, + bootstrapSupplier, + ignored -> false + ); + } + + public void publish(Long userId, NotificationStreamEventResponse eventResponse) { + sseSubscriptionRegistry.publish(channelKey(userId), NOTIFICATION_EVENT_NAME, eventResponse, false); + } + + private String channelKey(Long userId) { + return "notification:" + userId; + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java index 3f5f6dc..a6478ca 100644 --- a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java +++ b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java @@ -75,10 +75,9 @@ public SseEmitter subscribe( }); emitter.onError(ignored -> remove(channelKey, emitter)); - T initialPayload = initialPayloadSupplier.get(); - boolean completeAfterInitial = completeAfterInitialPredicate.test(initialPayload); - try { + T initialPayload = initialPayloadSupplier.get(); + boolean completeAfterInitial = completeAfterInitialPredicate.test(initialPayload); emitter.send(SseEmitter.event().name("connected").data("connected")); emitter.send(SseEmitter.event().name(eventName).data(initialPayload)); if (completeAfterInitial) { diff --git a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java index 523ac62..27fd097 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java @@ -2,9 +2,13 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingIngestResponse; +import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingResponse; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationTargetType; +import com.jobdri.jobdri_api.domain.notification.entity.NotificationType; import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask; import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.FailureReason; import com.jobdri.jobdri_api.domain.jobposting.repository.JobPostingAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.notification.service.NotificationService; import com.jobdri.jobdri_api.domain.user.entity.User; import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException; import org.junit.jupiter.api.BeforeEach; @@ -20,6 +24,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -33,6 +38,9 @@ class JobPostingAsyncTaskServiceTest { @Mock private JobPostingAsyncSseService jobPostingAsyncSseService; + @Mock + private NotificationService notificationService; + private JobPostingAsyncTaskService jobPostingAsyncTaskService; @BeforeEach @@ -40,7 +48,8 @@ void setUp() { jobPostingAsyncTaskService = new JobPostingAsyncTaskService( jobPostingAsyncTaskRepository, new ObjectMapper(), - jobPostingAsyncSseService + jobPostingAsyncSseService, + notificationService ); } @@ -172,5 +181,72 @@ void markRetryScheduledDoesNothingWhenTaskIsTerminal() { assertThat(task.getError()).isEqualTo(originalError); assertThat(task.getRetryCount()).isEqualTo(originalRetryCount); verify(jobPostingAsyncSseService, never()).publish(any()); + verify(notificationService, never()).createNotification(any(), any(), any(), any(), any(), any(), any()); + } + + @Test + @DisplayName("성공 처리 시 알림 서비스를 호출한다") + void markSuccessCreatesNotification() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + JobPostingResponse saved = JobPostingResponse.builder() + .jobPostingId(123L) + .build(); + JobPostingIngestResponse result = new JobPostingIngestResponse(true, null, null, null, null, null, saved); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markSuccess(task.getTaskId(), result); + + verify(notificationService).createNotification( + eq(7L), + eq(NotificationType.JOB_POSTING_ASYNC_SUCCEEDED), + eq("채용 공고 작업이 완료되었습니다."), + eq("채용 공고 분석과 저장이 완료되었습니다."), + eq(NotificationTargetType.JOB_POSTING_RESULT), + eq("123"), + any() + ); + } + + @Test + @DisplayName("실패 처리 시 사용자에게는 일반화된 실패 알림을 보낸다") + void markFailedCreatesSanitizedNotification() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markFailed( + task.getTaskId(), + FailureReason.INTERNAL_ERROR, + "worker stack trace", + 1 + ); + + verify(notificationService).createNotification( + eq(7L), + eq(NotificationType.JOB_POSTING_ASYNC_FAILED), + eq("채용 공고 작업이 실패했습니다."), + eq("채용 공고 작업 처리 중 오류가 발생했습니다."), + eq(NotificationTargetType.JOB_POSTING_TASK), + eq(task.getTaskId()), + any() + ); + } + + @Test + @DisplayName("재시도 예약 중에는 알림을 생성하지 않는다") + void markRetryScheduledDoesNotCreateNotification() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markRetryScheduled( + task.getTaskId(), + FailureReason.INTERNAL_ERROR, + "retry scheduled", + 1 + ); + + verify(notificationService, never()).createNotification(any(), any(), any(), any(), any(), any(), any()); } }