diff --git a/backend/src/main/java/pl/polskalokalnie/campaign/CampaignSender.java b/backend/src/main/java/pl/polskalokalnie/campaign/CampaignSender.java
new file mode 100644
index 0000000..04f1c42
--- /dev/null
+++ b/backend/src/main/java/pl/polskalokalnie/campaign/CampaignSender.java
@@ -0,0 +1,203 @@
+package pl.polskalokalnie.campaign;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.domain.Limit;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.util.HtmlUtils;
+import pl.polskalokalnie.lead.Lead;
+import pl.polskalokalnie.lead.LeadRepository;
+import pl.polskalokalnie.mailconfig.MailConfigService;
+import pl.polskalokalnie.mailconfig.SmsGatewayConfig;
+import pl.polskalokalnie.mailconfig.SmtpConfig;
+import pl.polskalokalnie.send.EmailSender;
+import pl.polskalokalnie.send.SendResult;
+import pl.polskalokalnie.send.SmsSender;
+import pl.polskalokalnie.unsubscribe.UnsubscribeService;
+
+/**
+ * Wysylka kampanii z outboxu. Nic nie rusza sama z siebie - wysyla wylacznie to, co administrator
+ * zaplanowal w panelu, porcjami i w granicach dziennego limitu kampanii.
+ *
+ *
Tuz przed wyslaniem kazdy odbiorca jest sprawdzany ponownie: zgoda mogla zostac cofnieta,
+ * a rezygnacja przyjeta juz po zbudowaniu outboxu. Taki odbiorca dostaje status OPTED_OUT.
+ */
+@Component
+public class CampaignSender {
+
+ private static final Logger log = LoggerFactory.getLogger(CampaignSender.class);
+
+ /** Ile pozycji outboxu przetwarzamy na jeden przebieg - reszta poczeka minute. */
+ private static final int BATCH_SIZE = 25;
+
+ private final CampaignRepository campaignRepository;
+ private final CampaignRecipientRepository recipientRepository;
+ private final LeadRepository leadRepository;
+ private final MailConfigService mailConfigService;
+ private final EmailSender emailSender;
+ private final SmsSender smsSender;
+ private final UnsubscribeService unsubscribeService;
+ private final String baseUrl;
+
+ public CampaignSender(CampaignRepository campaignRepository,
+ CampaignRecipientRepository recipientRepository,
+ LeadRepository leadRepository,
+ MailConfigService mailConfigService,
+ EmailSender emailSender,
+ SmsSender smsSender,
+ UnsubscribeService unsubscribeService,
+ @Value("${app.public-base-url:http://localhost}") String baseUrl) {
+ this.campaignRepository = campaignRepository;
+ this.recipientRepository = recipientRepository;
+ this.leadRepository = leadRepository;
+ this.mailConfigService = mailConfigService;
+ this.emailSender = emailSender;
+ this.smsSender = smsSender;
+ this.unsubscribeService = unsubscribeService;
+ this.baseUrl = baseUrl == null ? "http://localhost" : baseUrl.replaceAll("/+$", "");
+ }
+
+ @Scheduled(fixedDelay = 60_000, initialDelay = 30_000)
+ @Transactional
+ public void processOutbox() {
+ Instant now = Instant.now();
+ campaignRepository.findByStatus(CampaignStatus.SCHEDULED).stream()
+ .filter(campaign -> campaign.getScheduledAt() == null || !campaign.getScheduledAt().isAfter(now))
+ .forEach(campaign -> {
+ campaign.setStatus(CampaignStatus.SENDING);
+ campaignRepository.save(campaign);
+ });
+
+ for (Campaign campaign : campaignRepository.findByStatus(CampaignStatus.SENDING)) {
+ try {
+ sendBatch(campaign);
+ } catch (RuntimeException ex) {
+ log.warn("Kampania {}: przebieg wysylki nie powiodl sie: {}", campaign.getId(), ex.getMessage());
+ }
+ }
+ }
+
+ private void sendBatch(Campaign campaign) {
+ int remainingToday = remainingToday(campaign);
+ if (remainingToday <= 0) {
+ return;
+ }
+ List batch = recipientRepository.findByCampaignIdAndStatusOrderByIdAsc(
+ campaign.getId(), RecipientStatus.QUEUED, Limit.of(Math.min(BATCH_SIZE, remainingToday)));
+ if (batch.isEmpty()) {
+ campaign.setStatus(CampaignStatus.COMPLETED);
+ campaignRepository.save(campaign);
+ return;
+ }
+
+ SmtpConfig smtp = campaign.getChannel() == Channel.EMAIL ? mailConfigService.getSmtp() : null;
+ SmsGatewayConfig sms = campaign.getChannel() == Channel.SMS ? mailConfigService.getSms() : null;
+ // Bez skonfigurowanej skrzynki (albo bramki SMS) kolejka czeka - inaczej caly outbox
+ // spalilby sie na bledach, a kampanii nie da sie wyslac drugi raz.
+ if (!senderConfigured(smtp, sms)) {
+ log.warn("Kampania {}: brak konfiguracji kanalu {} - wysylka czeka.", campaign.getId(), campaign.getChannel());
+ return;
+ }
+ int sent = 0;
+
+ for (CampaignRecipient recipient : batch) {
+ if (stillOptedOut(recipient)) {
+ recipient.setStatus(RecipientStatus.OPTED_OUT);
+ recipientRepository.save(recipient);
+ continue;
+ }
+ SendResult result = campaign.getChannel() == Channel.EMAIL
+ ? emailSender.send(smtp, recipient.getAddress(), campaign.getSubject(), emailBody(campaign, recipient))
+ : smsSender.send(sms, recipient.getAddress(), smsBody(campaign, recipient));
+ if (result.ok()) {
+ recipient.setStatus(RecipientStatus.SENT);
+ recipient.setSentAt(Instant.now());
+ recipient.setProviderMessageId(result.providerMessageId());
+ recipient.setError(null);
+ sent++;
+ } else {
+ recipient.setStatus(RecipientStatus.FAILED);
+ recipient.setError(trim(result.error()));
+ }
+ recipientRepository.save(recipient);
+ }
+
+ campaign.setSentToday(todaysCount(campaign) + sent);
+ campaign.setSentTodayDate(LocalDate.now());
+ if (!recipientRepository.findByCampaignIdAndStatusOrderByIdAsc(campaign.getId(), RecipientStatus.QUEUED, Limit.of(1)).isEmpty()) {
+ campaignRepository.save(campaign);
+ return;
+ }
+ campaign.setStatus(CampaignStatus.COMPLETED);
+ campaignRepository.save(campaign);
+ }
+
+ static boolean senderConfigured(SmtpConfig smtp, SmsGatewayConfig sms) {
+ if (smtp != null) {
+ return smtp.isEnabled() && smtp.getHost() != null && !smtp.getHost().isBlank();
+ }
+ return sms != null && sms.isEnabled() && sms.getEndpointUrl() != null && !sms.getEndpointUrl().isBlank();
+ }
+
+ /** Ile jeszcze mozna dzis wyslac w tej kampanii - licznik zeruje sie z nowym dniem. */
+ private int remainingToday(Campaign campaign) {
+ int limit = campaign.getDailyLimit() == null || campaign.getDailyLimit() <= 0 ? Integer.MAX_VALUE : campaign.getDailyLimit();
+ return limit - todaysCount(campaign);
+ }
+
+ private int todaysCount(Campaign campaign) {
+ boolean isToday = LocalDate.now().equals(campaign.getSentTodayDate());
+ return isToday && campaign.getSentToday() != null ? campaign.getSentToday() : 0;
+ }
+
+ /** Zgoda mogla zostac cofnieta po zbudowaniu outboxu - sprawdzamy ja tuz przed wyslaniem. */
+ private boolean stillOptedOut(CampaignRecipient recipient) {
+ Lead lead = recipient.getLeadId() == null ? null : leadRepository.findById(recipient.getLeadId()).orElse(null);
+ if (lead == null) {
+ return false;
+ }
+ return recipient.getChannel() == Channel.EMAIL
+ ? !lead.isEmailConsent() || lead.isEmailOptOut()
+ : !lead.isSmsConsent() || lead.isSmsOptOut();
+ }
+
+ /** Kazda wiadomosc marketingowa musi dac sie wypisac jednym kliknieciem. */
+ private String unsubscribeLink(CampaignRecipient recipient) {
+ String channel = recipient.getChannel() == Channel.EMAIL ? "email" : "sms";
+ return baseUrl + "/api/unsubscribe?token=" + unsubscribeService.generateToken(recipient.getLeadId(), channel);
+ }
+
+ private String emailBody(Campaign campaign, CampaignRecipient recipient) {
+ String body = campaign.getBody() == null ? "" : campaign.getBody();
+ String greeting = recipient.getName() == null || recipient.getName().isBlank() ? ""
+ : "Cześć " + HtmlUtils.htmlEscape(recipient.getName().trim().split("\\s+")[0]) + ",
";
+ String link = recipient.getLeadId() == null ? null : unsubscribeLink(recipient);
+ String footer = link == null ? ""
+ : "
"
+ + "Otrzymujesz tę wiadomość, bo zgodziłeś(-aś) się na"
+ + " informacje marketingowe od serwisu Polska Lokalnie. "
+ + "Zrezygnuj z tych wiadomości.
";
+ return ""
+ + ""
+ + greeting + body + footer + "
";
+ }
+
+ private String smsBody(Campaign campaign, CampaignRecipient recipient) {
+ String body = campaign.getBody() == null ? "" : campaign.getBody().trim();
+ String link = recipient.getLeadId() == null ? null : unsubscribeLink(recipient);
+ return link == null ? body : body + " Rezygnacja: " + link;
+ }
+
+ private static String trim(String value) {
+ if (value == null) {
+ return null;
+ }
+ return value.length() > 240 ? value.substring(0, 240) : value;
+ }
+}
diff --git a/backend/src/main/java/pl/polskalokalnie/campaign/CampaignService.java b/backend/src/main/java/pl/polskalokalnie/campaign/CampaignService.java
index 081d78c..a54c9c1 100644
--- a/backend/src/main/java/pl/polskalokalnie/campaign/CampaignService.java
+++ b/backend/src/main/java/pl/polskalokalnie/campaign/CampaignService.java
@@ -85,8 +85,8 @@ public class CampaignService {
/**
* Uruchamia kampanie: buduje outbox (jesli jeszcze nie istnieje) i ustawia status. Kampania
- * z data w przyszlosci przechodzi w SCHEDULED, w przeciwnym razie od razu w SENDING - workera
- * wysylajacego dodajemy w kolejnej fazie.
+ * z data w przyszlosci przechodzi w SCHEDULED, w przeciwnym razie od razu w SENDING.
+ * Wysylke z outboxu obsluguje {@link CampaignSender}.
*/
@Transactional
public CampaignResponse schedule(Long id) {
diff --git a/backend/src/test/java/pl/polskalokalnie/campaign/CampaignSenderTest.java b/backend/src/test/java/pl/polskalokalnie/campaign/CampaignSenderTest.java
new file mode 100644
index 0000000..3df78bb
--- /dev/null
+++ b/backend/src/test/java/pl/polskalokalnie/campaign/CampaignSenderTest.java
@@ -0,0 +1,140 @@
+package pl.polskalokalnie.campaign;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Optional;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.domain.Limit;
+import pl.polskalokalnie.lead.Lead;
+import pl.polskalokalnie.lead.LeadRepository;
+import pl.polskalokalnie.mailconfig.MailConfigService;
+import pl.polskalokalnie.mailconfig.SmtpConfig;
+import pl.polskalokalnie.send.EmailSender;
+import pl.polskalokalnie.send.SendResult;
+import pl.polskalokalnie.send.SmsSender;
+import pl.polskalokalnie.unsubscribe.UnsubscribeService;
+
+class CampaignSenderTest {
+
+ private CampaignRepository campaignRepository;
+ private CampaignRecipientRepository recipientRepository;
+ private LeadRepository leadRepository;
+ private EmailSender emailSender;
+ private CampaignSender sender;
+
+ private Campaign campaign;
+
+ @BeforeEach
+ void setUp() {
+ campaignRepository = mock(CampaignRepository.class);
+ recipientRepository = mock(CampaignRecipientRepository.class);
+ leadRepository = mock(LeadRepository.class);
+ emailSender = mock(EmailSender.class);
+ MailConfigService mailConfigService = mock(MailConfigService.class);
+ UnsubscribeService unsubscribeService = mock(UnsubscribeService.class);
+
+ SmtpConfig smtp = new SmtpConfig();
+ smtp.setEnabled(true);
+ smtp.setHost("smtp.example.com");
+ when(mailConfigService.getSmtp()).thenReturn(smtp);
+ when(unsubscribeService.generateToken(any(), anyString())).thenReturn("token");
+ when(emailSender.send(any(), anyString(), any(), anyString())).thenReturn(new SendResult(true, null, null));
+
+ campaign = new Campaign();
+ campaign.setChannel(Channel.EMAIL);
+ campaign.setSubject("Nowe oferty");
+ campaign.setBody("Treść
");
+ campaign.setDailyLimit(200);
+ campaign.setStatus(CampaignStatus.SENDING);
+ when(campaignRepository.findByStatus(CampaignStatus.SCHEDULED)).thenReturn(List.of());
+ when(campaignRepository.findByStatus(CampaignStatus.SENDING)).thenReturn(List.of(campaign));
+
+ sender = new CampaignSender(campaignRepository, recipientRepository, leadRepository, mailConfigService,
+ emailSender, mock(SmsSender.class), unsubscribeService, "https://polskalokalnie.pl");
+ }
+
+ private CampaignRecipient recipient(long leadId, String address) {
+ CampaignRecipient recipient = new CampaignRecipient();
+ recipient.setLeadId(leadId);
+ recipient.setChannel(Channel.EMAIL);
+ recipient.setAddress(address);
+ recipient.setStatus(RecipientStatus.QUEUED);
+ return recipient;
+ }
+
+ private Lead lead(boolean consent, boolean optOut) {
+ Lead lead = new Lead();
+ lead.setEmailConsent(consent);
+ lead.setEmailOptOut(optOut);
+ return lead;
+ }
+
+ private void queue(List batch) {
+ when(recipientRepository.findByCampaignIdAndStatusOrderByIdAsc(any(), eq(RecipientStatus.QUEUED), any(Limit.class)))
+ .thenReturn(batch, List.of());
+ }
+
+ @Test
+ void sendsQueuedRecipientAndClosesCampaign() {
+ CampaignRecipient recipient = recipient(1L, "kupujacy@example.com");
+ queue(List.of(recipient));
+ when(leadRepository.findById(1L)).thenReturn(Optional.of(lead(true, false)));
+
+ sender.processOutbox();
+
+ assertThat(recipient.getStatus()).isEqualTo(RecipientStatus.SENT);
+ assertThat(recipient.getSentAt()).isNotNull();
+ // Pusta kolejka po przebiegu = kampania rozliczona.
+ assertThat(campaign.getStatus()).isEqualTo(CampaignStatus.COMPLETED);
+ assertThat(campaign.getSentToday()).isEqualTo(1);
+ }
+
+ @Test
+ void skipsRecipientWhoWithdrewConsentAfterBuildingOutbox() {
+ CampaignRecipient recipient = recipient(2L, "rezygnacja@example.com");
+ queue(List.of(recipient));
+ when(leadRepository.findById(2L)).thenReturn(Optional.of(lead(true, true)));
+
+ sender.processOutbox();
+
+ assertThat(recipient.getStatus()).isEqualTo(RecipientStatus.OPTED_OUT);
+ verify(emailSender, never()).send(any(), anyString(), any(), anyString());
+ }
+
+ @Test
+ void marksFailureWithoutLosingTheRest() {
+ CampaignRecipient recipient = recipient(3L, "blad@example.com");
+ queue(List.of(recipient));
+ when(leadRepository.findById(3L)).thenReturn(Optional.of(lead(true, false)));
+ when(emailSender.send(any(), anyString(), any(), anyString()))
+ .thenReturn(new SendResult(false, null, "Nie udało się połączyć"));
+
+ sender.processOutbox();
+
+ assertThat(recipient.getStatus()).isEqualTo(RecipientStatus.FAILED);
+ assertThat(recipient.getError()).contains("połączyć");
+ assertThat(campaign.getSentToday()).isZero();
+ }
+
+ @Test
+ void waitsWhenDailyLimitIsUsedUp() {
+ campaign.setDailyLimit(1);
+ campaign.setSentToday(1);
+ campaign.setSentTodayDate(java.time.LocalDate.now());
+
+ sender.processOutbox();
+
+ verify(recipientRepository, never())
+ .findByCampaignIdAndStatusOrderByIdAsc(any(), eq(RecipientStatus.QUEUED), any(Limit.class));
+ assertThat(campaign.getStatus()).isEqualTo(CampaignStatus.SENDING);
+ }
+}