File: /home/xedaptot/ai.naniguide.com/app/Model/Campaign.php
<?php
/**
* Campaign class.
*
* Model class for campaigns related functionalities.
* This is the center of the application
*
* LICENSE: This product includes software developed at
* the Acelle Co., Ltd. (http://acellemail.com/).
*
* @category MVC Model
*
* @copyright Acelle Co., Ltd
* @license Acelle Co., Ltd
*
* @version 1.0
*
* @link http://acellemail.com
*/
namespace App\Model;
use DB;
use App\Model\SendingServer;
use Carbon\Carbon;
use League\Csv\Writer;
use App\Library\StringHelper;
use App\Model\Setting;
use Validator;
use ZipArchive;
use Exception;
use App\Library\Traits\HasEmailTemplate;
use App\Library\RouletteWheel;
use Throwable;
use App\Jobs\ExecuteCampaignCallback;
use App\Library\HtmlHandler\InjectTrackingPixel;
use App\Library\HtmlHandler\TransformUrl;
use Illuminate\Database\Eloquent\Model;
use App\Library\Traits\HasUid;
use App\Library\Traits\HasModelCacheIdentity;
use App\Library\Traits\TrackJobs;
use App\Library\Cache\AppCache;
use App\Library\Cache\Cacheable;
use Monolog\Logger as MonologLogger;
use Monolog\Handler\RotatingFileHandler;
use Monolog\Formatter\LineFormatter;
use App\Jobs\LoadCampaign;
use App\Jobs\RunCampaign;
use App\Model\AbTest;
use App\Model\AbTestVariant;
use App\Jobs\EvaluateAbTestWinner;
use App\Jobs\HandleDuplicateEmails;
use Illuminate\Bus\Batch;
use App\Events\CampaignUpdated;
use Illuminate\Database\Eloquent\Factories\HasFactory;
use Illuminate\Support\Facades\Cache;
use App\Library\Contracts\CampaignInterface;
use Closure;
use App\Library\Contracts\TagResolverInterface;
use App\Services\TemplateService;
use App\Services\Plans\Entitlements\EntitlementKey;
use App\Services\Plans\Gates\EntitlementGate;
/**
* @property-read \App\Model\Customer $customer
*/
class Campaign extends Model implements CampaignInterface, TagResolverInterface, Cacheable
{
use HasEmailTemplate;
use TrackJobs;
use HasUid;
use HasModelCacheIdentity;
use HasFactory;
protected $logger;
/**
* Content fields delegated to the `email()` relation when `email_id` is set.
* Reads/writes to these attributes transparently go through Email.
* Existing code using $this->subject etc. continues to work unchanged.
*/
protected const EMAIL_ATTRIBUTES = [
'type',
'subject', 'from_email', 'from_name', 'reply_to',
'html', 'plain', 'sign_dkim', 'track_open', 'track_click',
'use_default_sending_server_from_email', 'skip_failed_message',
'preheader', 'delivery_statuses', 'template_id',
'tracking_domain_id', 'signature_id',
];
public function getAttribute($key)
{
if (in_array($key, static::EMAIL_ATTRIBUTES) && !is_null($this->email_id)) {
return optional($this->getRelationValue('email') ?? $this->email)->getAttribute($key);
}
return parent::getAttribute($key);
}
/**
* Delegate $campaign->template to email->template relation.
*/
public function getTemplateAttribute()
{
return $this->email?->template;
}
public function setLastError($message, $backtrace = null): self
{
if (!is_string($message)) {
throw new \InvalidArgumentException('Campaign last_error message must be a string and cannot be null.');
}
if (!is_null($backtrace) && !is_string($backtrace)) {
throw new \InvalidArgumentException('Campaign last_error backtrace must be a string or null.');
}
$message = trim($message);
$backtrace = is_null($backtrace) ? '' : trim($backtrace);
if ($message === '') {
throw new \InvalidArgumentException('Campaign last_error message cannot be empty.');
}
$this->attributes['last_error'] = json_encode([
'message' => $message,
'backtrace' => $backtrace,
], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR);
return $this;
}
public function setLastErrorAttribute($value): void
{
if (!is_null($value) && !is_string($value)) {
throw new \InvalidArgumentException(
'Campaign last_error attribute only accepts null or string. Use setLastError($message, $backtrace = null) for structured errors.'
);
}
if (is_null($value)) {
$this->attributes['last_error'] = null;
return;
}
$this->setLastError($value);
}
public function getLastErrorAttribute($value): ?string
{
if (is_null($value) || $value === '') {
return null;
}
// Read accessor must be graceful — campaign listing renders this for every
// row, so any malformed legacy/raw value would crash the entire page.
// Strict validation belongs on write (setLastError), not read.
try {
$payload = json_decode($value, true, 512, JSON_THROW_ON_ERROR);
} catch (\JsonException $e) {
// Legacy plain-string error → return as-is, trimmed
return trim($value) !== '' ? trim($value) : null;
}
if (!is_array($payload)) {
return is_scalar($payload) ? (string) $payload : null;
}
$message = isset($payload['message']) && is_string($payload['message'])
? trim($payload['message'])
: '';
return $message !== '' ? $message : null;
}
public function setAttribute($key, $value)
{
if (in_array($key, static::EMAIL_ATTRIBUTES) && !is_null($this->email_id)) {
$email = $this->getRelationValue('email') ?? $this->email;
if ($email) {
$email->setAttribute($key, $value);
return $this;
}
}
return parent::setAttribute($key, $value);
}
// 4 types of delivery status for a given contact
public const DELIVERY_STATUS_FAILED = 'failed';
public const DELIVERY_STATUS_SENT = 'sent';
public const DELIVERY_STATUS_NEW = 'new';
public const DELIVERY_STATUS_SKIPPED = 'skipped';
public const DELIVERY_STATUS_BOUNCED = 'bounced';
public const DELIVERY_STATUS_FEEDBACK = 'feedback';
public const DELIVERY_STATUS_UNSUBSCRIBED = 'unsubscribed';
protected $serversPool = null;
protected $casts = [
'created_at' => 'datetime',
'updated_at' => 'datetime',
'run_at' => 'datetime',
'delivery_at' => 'datetime',
'is_error' => 'boolean',
'is_paused' => 'boolean',
];
/**
* Get campaign's default mail list.
*/
public function defaultMailList()
{
return $this->belongsTo('App\Model\MailList', 'default_mail_list_id');
}
/**
* Get campaign's associated mail list.
*/
public function mailLists()
{
return $this->belongsToMany('App\Model\MailList', 'campaigns_lists_segments');
}
/**
* Per-campaign sending server narrowing. Empty = use full pool from
* MailList::getSendingServers() (default broadcast behavior).
*/
public function sendingServersPivot()
{
return $this->hasMany(CampaignSendingServer::class, 'campaign_id');
}
/**
* Campaign has many campaign links.
*/
public function campaignLinks()
{
return $this->hasMany('App\Model\CampaignLink');
}
public function campaignArchives()
{
return $this->hasMany(CampaignArchive::class);
}
/**
* Campaign has many campaign webhooks.
*/
public function campaignWebhooks()
{
return $this->hasMany('App\Model\CampaignWebhook');
}
public function campaignHeaders()
{
return $this->hasMany('App\Model\CampaignHeader');
}
/**
* Campaign has no `tracking_domain_id` column. The FK lives on the `email`
* row (see EMAIL_ATTRIBUTES proxy). Expose `$campaign->trackingDomain` as
* an accessor that delegates to the Email's belongsTo relation.
*/
public function getTrackingDomainAttribute(): ?\App\Model\TrackingDomain
{
return $this->email?->trackingDomain;
}
public function signature()
{
return $this->belongsTo('App\Model\Signature', 'signature_id');
}
public function setSignature($signature)
{
$this->signature_id = $signature ? $signature->id : null;
$this->save();
}
/**
* Association with attachments.
*/
public function attachments()
{
return $this->hasMany('App\Model\Attachment', 'campaign_id');
}
/**
* Get campaign validation rules.
*/
public function rules($request = null)
{
$rules = array(
'name' => 'required',
'subject' => 'required',
'from_email' => 'required|email',
'from_name' => 'required',
'reply_to' => 'required|email',
);
if ($this->use_default_sending_server_from_email) {
$rules['from_email'] = ['nullable', 'email'];
$rules['reply_to'] = ['nullable', 'email'];
} else {
$rules['from_email'] = ['required', 'email'];
$rules['reply_to'] = ['required', 'email'];
}
// Backend enforcement: from_email must be in the customer's visible
// verified-senders list — same list the UI's `_identity_select` AJAX
// dropdown shows (powered by `IdentityCatalog::valuesForCustomer`).
// Don't trust the client: even though the wizard's combobox restricts
// the user to verified picks, an attacker (or buggy edit-resume flow)
// can POST any from_email; vendor reject at send time gives a worse
// failure mode than a friendly form error here.
//
// Skipped when the customer is allowed unverified senders (per the
// sending server's `allow_unverified_from_email` option /
// `custom.sign_with_default_domain` global), or when
// `use_default_sending_server_from_email` is on (the server's own
// default is substituted, from_email is just metadata).
$customer = $this->customer;
if ($customer
&& !$this->use_default_sending_server_from_email
&& !$customer->allowUnverifiedFromEmailAddress()) {
$rules['from_email'][] = function ($attribute, $value, $fail) use ($customer) {
if (empty($value)) {
return;
}
$emailOnly = trim($value);
if (preg_match('/<([^>]+)>\s*$/', $emailOnly, $m)) {
$emailOnly = trim($m[1]);
}
$visible = app(\App\SendingServers\Identities\IdentityCatalog::class)
->valuesForCustomer($customer);
// valuesForCustomer returns mixed shapes — bare emails AND
// "Name <email>" forms (from own Sender records). Normalize
// to lowercase emails for the comparison.
$visibleEmails = [];
foreach ($visible as $entry) {
if (preg_match('/<([^>]+)>\s*$/', $entry, $mm)) {
$visibleEmails[] = strtolower(trim($mm[1]));
} else {
$visibleEmails[] = strtolower(trim($entry));
}
}
if (!in_array(strtolower($emailOnly), $visibleEmails, true)) {
$fail(trans('refactor/campaigns.wizard.identity_error', [
'sender_link' => route('refactor.sending.senders'),
]));
}
};
}
// tracking domain
if (isset($request) && $request->custom_tracking_domain) {
$rules['tracking_domain_uid'] = 'required';
}
// Per-campaign sending-server narrow.
// Two states distinguishable by request shape:
// 1. Master toggle "Use all servers" ON → JS marks every <input> in
// the narrow block `disabled`, so `sending_servers` is missing
// entirely. Pivot gets cleared. (nullable path)
// 2. Master toggle OFF → at minimum the hidden `[fitness]` inputs
// post, so `sending_servers` IS present in the request. At least
// one row must carry a real `sending_server_id`, otherwise the
// customer narrowed to nothing — fail loud rather than silently
// fall back to "use all".
// CampaignSetupDto::fromRequest filters phantom rows (fitness without
// sending_server_id), so per-row nullable is fine; the integrity check
// is the closure below.
$rules['sending_servers'] = ['nullable', 'array', function ($attribute, $value, $fail) {
if (! is_array($value)) {
return;
}
$hasAny = false;
foreach ($value as $row) {
if (is_array($row) && !empty($row['sending_server_id'])) {
$hasAny = true;
break;
}
}
if (! $hasAny) {
$fail(trans('refactor/campaigns.wizard.error_no_servers_selected'));
}
}];
$rules['sending_servers.*.sending_server_id'] = [
'nullable',
'integer',
\Illuminate\Validation\Rule::in($this->availableSendingServerIds()),
];
$rules['sending_servers.*.fitness'] = 'nullable|integer|min:1|max:100';
return $rules;
}
/**
* Get campaign tracking logs.
*
* @return mixed
*/
public function trackingLogs()
{
return $this->hasMany('App\Model\TrackingLog');
}
/**
* Get campaign bounce logs.
*
* @return mixed
*/
public function bounceLogs()
{
return BounceLog::select('bounce_logs.*')->leftJoin('tracking_logs', 'tracking_logs.id', '=', 'bounce_logs.tracking_log_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Get campaign open logs.
*
* @return mixed
*/
public function openLogs()
{
return OpenLog::select('open_logs.*')->leftJoin('tracking_logs', 'tracking_logs.message_id', '=', 'open_logs.message_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Get campaign click logs.
*
* @return mixed
*/
public function clickLogs()
{
return ClickLog::select('click_logs.*')->leftJoin('tracking_logs', 'tracking_logs.message_id', '=', 'click_logs.message_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Get campaign feedback loop logs.
*
* @return mixed
*/
public function feedbackLogs()
{
return FeedbackLog::select('feedback_logs.*')->leftJoin('tracking_logs', 'tracking_logs.id', '=', 'feedback_logs.tracking_log_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Get campaign unsubscribe logs.
*
* @return mixed
*/
public function unsubscribeLogs()
{
return UnsubscribeLog::select('unsubscribe_logs.*')->leftJoin('tracking_logs', 'tracking_logs.message_id', '=', 'unsubscribe_logs.message_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Get campaign list segment.
*
* @return \Illuminate\Database\Eloquent\Relations\HasMany<CampaignsListsSegment, $this>
*/
public function listsSegments()
{
return $this->hasMany(CampaignsListsSegment::class);
}
/**
* Get campaign lists segments.
*
* @return mixed
*/
public function getListsSegments()
{
$lists_segments = $this->listsSegments;
if ($lists_segments->isEmpty()) {
$lists_segment = new CampaignsListsSegment();
$lists_segment->customer_id = $this->customer_id;
$lists_segment->campaign_id = $this->id;
$lists_segment->is_default = true;
$lists_segments->push($lists_segment);
}
return $lists_segments;
}
/**
* Get campaign lists segments group by list.
*
* @return mixed
*/
public function getListsSegmentsGroups()
{
$lists_segments = $this->getListsSegments();
$groups = [];
foreach ($lists_segments as $lists_segment) {
if (!isset($groups[$lists_segment->mail_list_id])) {
$groups[$lists_segment->mail_list_id] = [];
$groups[$lists_segment->mail_list_id]['list'] = $lists_segment->mailList;
if ($this->default_mail_list_id == $lists_segment->mail_list_id) {
$groups[$lists_segment->mail_list_id]['is_default'] = true;
} else {
$groups[$lists_segment->mail_list_id]['is_default'] = false;
}
$groups[$lists_segment->mail_list_id]['segment_uids'] = [];
}
if ($lists_segment->segment && !in_array($lists_segment->segment->uid, $groups[$lists_segment->mail_list_id]['segment_uids'])) {
$groups[$lists_segment->mail_list_id]['segment_uids'][] = $lists_segment->segment->uid;
}
}
return $groups;
}
/**
* Check if the campaign setting is "use sending server's FROM email address".
*
* @return mixed
*/
private function useSendingServerFromEmailAddress()
{
return $this->use_default_sending_server_from_email == true;
}
/**
* Reset max_execution_time so that command can run for a long time without being terminated.
*
* @return mixed
*/
public static function resetMaxExecutionTime()
{
set_time_limit(0);
ini_set('max_execution_time', 0);
ini_set('memory_limit', '-1');
}
/**
* Log delivery message, used for later tracking.
*/
public function trackMessage(
\App\SendingServers\Drivers\SendResult $response,
$subscriber,
$server,
$msgId,
$triggerId = null,
?int $abTestVariantId = null
) {
$runtimeMessageId = $response->getRuntimeMessageId() ?? $msgId;
if ($response->getRuntimeMessageId() !== null) {
Customer::mapKeyToCustomer('runtime_message_id', $runtimeMessageId, $this->customer->id);
}
$params = [
'message_id' => $msgId,
'subscriber_id' => $subscriber->id,
'sending_server_id' => $server->id,
'customer_id' => $this->customer->id,
'status' => $response->getStatus()->value,
'runtime_message_id' => $runtimeMessageId,
'error' => $response->getError(),
];
if ($abTestVariantId !== null) {
$params['ab_test_variant_id'] = $abTestVariantId;
}
// create tracking log for message
$this->setConnection($this->customer->getDbConnection())->trackingLogs()->create($params);
}
/**
* The attributes that are mass assignable.
*
* @var array
*/
protected $fillable = [
'name', 'email_id',
// EMAIL_ATTRIBUTES are kept fillable so fill() passes them through to
// setAttribute(), which delegates them to the Email model.
'type', 'subject', 'from_email', 'from_name', 'reply_to',
'html', 'plain', 'sign_dkim', 'track_open', 'track_click',
'use_default_sending_server_from_email', 'skip_failed_message',
'preheader', 'delivery_statuses', 'template_id',
'tracking_domain_id', 'signature_id',
];
/**
* The rules for validation.
*
* @var array
*/
public static $rules = array(
'mail_list_uid' => 'required',
);
/**
* Items per page.
*
* @var array
*/
public static $itemsPerPage = 25;
/**
* Get all items.
*
* @return collect
*/
public static function getAll()
{
return self::select('campaigns.*');
}
/**
* Get select options.
*
* @return array
*/
public static function getSelectOptions($customer = null, $status = null)
{
$query = self::getAll();
if ($customer) {
$query = $query->where('customer_id', '=', $customer->id);
}
if (isset($status)) {
$query = $query->where('status', '=', $status);
}
$options = $query->orderBy('created_at', 'DESC')->get()->map(function ($item) {
return ['value' => $item->uid, 'text' => $item->name];
});
return $options;
}
/**
* Get current links of campaign.
*/
public function getLinks()
{
return $this->campaignLinks()->get();
}
/**
* Get urls from campaign html.
*/
public function getUrls()
{
// Find all links in campaign content
preg_match_all('/<a[^>]*href=["\'](?<url>http[^"\']*)["\']/i', $this->getTemplateContent(), $matches);
$hrefs = array_unique($matches['url']);
$urls = [];
foreach ($hrefs as $href) {
if (preg_match('/^http/i', $href) && !\App\Library\StringHelper::containsUnsubscribeTag($href)) {
$urls[] = strtolower(trim($href));
}
}
return $urls;
}
/**
* Get all URLs from A/B test variant emails.
*/
public function getAbTestUrls(): array
{
$allUrls = [];
foreach ($this->abTest->variants()->with('email')->get() as $variant) {
if (!$variant->email) {
continue;
}
$content = $variant->email->getTemplateContent();
preg_match_all('/<a[^>]*href=["\'](?<url>http[^"\']*)["\']/i', $content, $matches);
foreach (array_unique($matches['url']) as $href) {
if (preg_match('/^http/i', $href) && !\App\Library\StringHelper::containsUnsubscribeTag($href)) {
$allUrls[] = strtolower(trim($href));
}
}
}
return array_unique($allUrls);
}
/**
* Update campaign links.
*/
public function updateLinks()
{
if ($this->type == Email::TYPE_PLAIN_TEXT) {
return;
}
$this->campaignLinks()->delete();
$urls = $this->hasAbTest()
? $this->getAbTestUrls()
: $this->getUrls();
foreach ($urls as $url) {
// Campaign link
if ($this->campaignLinks()->where('url', '=', $url)->count() == 0) {
$cl = new CampaignLink();
$cl->customer_id = $this->customer_id;
$cl->campaign_id = $this->id;
$cl->url = $url;
$cl->setConnection($this->customer->getDbConnection())->save();
}
}
}
/**
* CHeck UNSUBSCRIBE_URL.
*
* @return object
*/
public function unsubscribe_url_valid()
{
if ($this->type != 'plain-text' &&
\App\Services\Plans\Gates\EntitlementGate::allows(
$this->customer,
\App\Services\Plans\Entitlements\EntitlementKey::HAS_UNSUBSCRIBE_URL_REQUIRED
) &&
!\App\Library\StringHelper::containsUnsubscribeTag($this->getTemplateContent())
) {
return false;
} else {
return true;
}
}
/**
* True when all required Setup fields are filled (name, subject, from_name, from_email/reply_to).
*/
public function hasSetupFields(): bool
{
return !empty($this->name)
&& !empty($this->subject)
&& !empty($this->from_name)
&& ($this->use_default_sending_server_from_email
|| (!empty($this->from_email) && !empty($this->reply_to)));
}
/**
* Wizard completion step (0–5).
*
* 0 = nothing, 1 = recipients done, 2 = setup done,
* 3 = template done, 4 = schedule done, 5 = ready to launch.
*/
public function step(): int
{
// Step 1: recipients
if (!$this->defaultMailList) {
return 0;
}
// Step 2: setup fields
if (!$this->hasSetupFields()) {
return 1;
}
// Step 3: template content
if ($this->hasAbTest()) {
// A/B test: check that at least one variant has a template
$hasTemplate = $this->abTest->variants()
->whereHas('email', fn ($q) => $q->whereNotNull('template_id')->orWhere('html', '!=', ''))
->exists();
} else {
$hasTemplate = $this->type === 'plain-text'
? !empty($this->plain)
: ($this->email && ($this->email->template || !empty($this->email->html)));
}
if (!$hasTemplate) {
return 2;
}
// Step 4+5: schedule defaults to "send now" — always accessible once template is done
return 5;
}
public function scopeError($query)
{
return $query->where('is_error', true);
}
/**
* Filter items.
*
* @return collect
*/
public static function scopeFilter($query, $request = null)
{
if (is_null($request)) {
return $query;
}
// Get campaign from ... (all|normal|automated)
if ($request->source == 'template') {
$query = $query->whereHas('email', fn ($q) => $q->whereNotNull('html'));
}
// Status
if (!empty(trim($request->statuses))) {
$query = $query->whereIn('status', explode(',', $request->statuses));
}
}
/**
* Search items.
*
* @return collect
*/
public static function scopeSearch($query, $keyword)
{
// Keyword
if (!empty(trim($keyword))) {
$query = $query->where('name', 'like', '%'.$keyword.'%');
}
}
/**
* Count delivery processed.
*
* @return number
*/
public function deliveredCount()
{
// including bounced, feedbcak...
return $this->trackingLogs()->sent()->count();
}
/**
* Count failed processed.
*
* @return number
*/
public function failedCount()
{
return $this->trackingLogs()->failed()->count();
}
/**
* Count failed processed.
*
* @return number
*/
public function notDeliveredCount()
{
$subscribersCountUniq = $this->subscribers([])->count();
return $subscribersCountUniq - $this->deliveredCount();
}
/**
* Count delivery success rate.
*
* @return number
*/
public function deliveredRate($cache = false)
{
$total = $this->subscribersCount($cache);
if ($total == 0) {
return 0;
}
return $this->deliveredCount() / $total;
}
/**
* Count delivery success rate.
*
* @return number
*/
public function failedRate($cache = false)
{
$total = $this->subscribersCount($cache);
if ($total == 0) {
return 0;
}
return $this->failedCount() / $total;
}
/**
* Count delivery success rate.
*
* @return number
*/
public function notDeliveredRate($cache = false)
{
$total = $this->subscribersCount($cache);
if ($total == 0) {
return 0;
}
return $this->notDeliveredCount() / $total;
}
/**
* Count click.
*
* @return number
*/
public function clickCount($start = null, $end = null)
{
$query = $this->clickLogs();
if (isset($start)) {
$query = $query->where('click_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('click_logs.created_at', '<=', $end);
}
return $query->count();
}
/**
* Url count.
*
* @return number
*/
public function urlCount()
{
return $this->campaignLinks()->count();
}
/**
* Count unique clicked opened emails.
*
* @return number
*/
public function uniqueClickCount()
{
$query = $this->clickLogs();
return $query->distinct('subscriber_id')->count('subscriber_id');
}
/**
* Clicked emails count.
*
* @return number
*/
public function clickRate()
{
$deliveryCount = $this->deliveredCount();
if ($deliveryCount == 0) {
return 0;
}
return $this->uniqueClickCount() / $deliveryCount;
}
/**
* Count abuse feedback.
*
* @return number
*/
public function abuseFeedbackCount()
{
return $this->feedbackLogs()->where('feedback_type', '=', 'abuse')->count();
}
/**
* Count open.
*
* @return number
*/
public function uniqueOpenCount()
{
return $this->openLogs()->distinct('tracking_logs.subscriber_id')->count();
}
/**
* Not open count.
*
* @return number
*/
public function notOpenCount()
{
return $this->deliveredCount() - $this->uniqueOpenCount();
}
/**
* Count unique open.
*
* @return number
*/
public function openUniqCount($start = null, $end = null)
{
$query = $this->openLogs();
if (isset($start)) {
$query = $query->where('open_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('open_logs.created_at', '<=', $end);
}
return $query->distinct('subscriber_id')->count('subscriber_id');
}
/**
* Open rate.
*
* @return number
*/
public function openRate()
{
$deliveredCount = $this->deliveredCount();
if ($deliveredCount == 0) {
return 0;
}
return $this->uniqueOpenCount() / $deliveredCount;
}
/**
* Not open rate.
*
* @return number
*/
public function notOpenRate()
{
return 1.0 - $this->openRate();
}
/**
* Count bounce back.
*
* @return number
*/
public function feedbackCount()
{
return $this->feedbackLogs()->distinct('subscriber_id')->count('subscriber_id');
}
/**
* feedbackRate.
*
* @return number
*/
public function feedbackRate()
{
$deliveredCount = $this->deliveredCount();
if ($deliveredCount == 0) {
return 0;
}
return $this->feedbackCount() / $deliveredCount;
}
public function bounceCount()
{
return $this->bounceLogs()->distinct('subscriber_id')->count('subscriber_id');
}
/**
* Count bounce rate.
*
* @return number
*/
public function bounceRate()
{
$deliveredCount = $this->deliveredCount();
if ($deliveredCount == 0) {
return 0;
}
return $this->bounceCount() / $deliveredCount;
}
/**
* Count unsubscibe.
*
* @return number
*/
public function unsubscribeCount()
{
return $this->unsubscribeLogs()->distinct('unsubscribe_logs.subscriber_id')->count();
}
/**
* Count unsubscibe rate.
*
* @return number
*/
public function unsubscribeRate()
{
$deliveredCount = $this->deliveredCount();
if ($deliveredCount == 0) {
return 0;
}
return $this->unsubscribeCount() / $deliveredCount;
}
/**
* Get last click.
*
* @param number $number
*
* @return collect
*/
public function lastClick()
{
return $this->clickLogs()->orderBy('created_at', 'desc')->first();
}
/**
* Get last open.
*
* @param number $number
*
* @return collect
*/
public function lastOpen()
{
return $this->openLogs()->orderBy('created_at', 'desc')->first();
}
/**
* Get last open list.
*
* @param number $number
*
* @return collect
*/
public function lastOpens($number)
{
return $this->openLogs()->orderBy('created_at', 'desc')->limit($number);
}
/**
* Get last opened time.
*
* @return \Carbon\Carbon|null
*/
public function getLastOpen()
{
$last = $this->campaign_track_opens()->orderBy('created_at', 'desc')->first();
return $last ? $last->created_at : null;
}
public static function topOpens($number = 5, $customer = null)
{
$records = self::select('campaigns.name', 'campaigns.id', 'campaigns.uid')
->addSelect(DB::raw('count(*) as aggregate'))
->join('tracking_logs', 'tracking_logs.campaign_id', '=', 'campaigns.id')
->join('open_logs', 'open_logs.message_id', '=', 'tracking_logs.message_id');
if (isset($customer)) {
$records = $records->where('campaigns.customer_id', '=', $customer->id);
}
$records = $records->groupBy('campaigns.name', 'campaigns.id', 'campaigns.uid')
->orderBy('aggregate', 'desc');
return $records->take($number);
}
public static function topClicks($number = 5, $customer = null)
{
$records = self::select('campaigns.name', 'campaigns.id', 'campaigns.uid')
->addSelect(DB::raw('count(*) as aggregate'))
->join('tracking_logs', 'tracking_logs.campaign_id', '=', 'campaigns.id')
->join('click_logs', 'click_logs.message_id', '=', 'tracking_logs.message_id');
if (isset($customer)) {
$records = $records->where('campaigns.customer_id', '=', $customer->id);
}
$records = $records->groupBy('campaigns.name', 'campaigns.id', 'campaigns.uid')
->orderBy('aggregate', 'desc');
return $records->take($number);
}
public static function topLinks($number = 5, $customer = null)
{
$records = CampaignLink::select('campaign_links.url')
->addSelect(DB::raw('count(*) as aggregate'))
->join('tracking_logs', 'tracking_logs.campaign_id', '=', 'campaign_links.campaign_id')
->join('click_logs', function ($join) {
$join->on('click_logs.message_id', '=', 'tracking_logs.message_id')
->on('click_logs.url', '=', 'campaign_links.url');
});
if (isset($customer)) {
$records = $records->join('campaigns', 'campaign_links.campaign_id', '=', 'campaigns.id')
->where('campaigns.customer_id', '=', $customer->id);
}
$records = $records->groupBy('campaign_links.url')
->orderBy('aggregate', 'desc');
return $records->take($number);
}
/**
* Campaign top 5 clicks.
*
* @return \Illuminate\Database\Eloquent\Builder
*/
public function getTopLinks($number = 5)
{
$records = $this->clickLogs()
->select('click_logs.url')
->addSelect(DB::raw('count(*) as aggregate'))
->groupBy('click_logs.url');
return $records->take($number);
}
/**
* Top subscribers by combined engagement (opens + clicks) for this campaign.
*
* Returns each subscriber's id/uid/email plus per-subscriber `opens` and
* `clicks` counts (both fields populated as integers, including 0 for
* subscribers who only opened or only clicked). Ordered by opens+clicks
* sum descending, with opens/clicks individually as tiebreakers.
*
* Uses two `tracking_log_id` LEFT-JOIN subqueries (instead of a single
* cross-join + DISTINCT) to keep the result row count linear: a
* subscriber with 12 opens × 7 clicks would balloon to 84 cartesian rows
* before COUNT(DISTINCT) collapsed it. Subqueries pre-aggregate.
*
* @return \Illuminate\Database\Query\Builder
*/
public function getTopActiveSubscribers($number = 5)
{
$opensSub = DB::table('open_logs as ol')
->select('t.subscriber_id', DB::raw('COUNT(*) AS opens'))
->join('tracking_logs as t', 't.id', '=', 'ol.tracking_log_id')
->where('t.campaign_id', $this->id)
->groupBy('t.subscriber_id');
$clicksSub = DB::table('click_logs as cl')
->select('t.subscriber_id', DB::raw('COUNT(*) AS clicks'))
->join('tracking_logs as t', 't.id', '=', 'cl.tracking_log_id')
->where('t.campaign_id', $this->id)
->groupBy('t.subscriber_id');
return DB::table('subscribers')
->select('subscribers.id', 'subscribers.uid', 'subscribers.email')
->selectRaw('COALESCE(o.opens, 0) AS opens')
->selectRaw('COALESCE(c.clicks, 0) AS clicks')
->leftJoinSub($opensSub, 'o', 'o.subscriber_id', '=', 'subscribers.id')
->leftJoinSub($clicksSub, 'c', 'c.subscriber_id', '=', 'subscribers.id')
->whereRaw('(COALESCE(o.opens, 0) + COALESCE(c.clicks, 0)) > 0')
->orderByRaw('(COALESCE(o.opens, 0) + COALESCE(c.clicks, 0)) DESC')
->orderByDesc('opens')
->orderByDesc('clicks')
->when($number !== null, fn ($q) => $q->limit($number));
}
public function getTopActiveSubscribersToCsv($file)
{
$records = $this->getTopActiveSubscribers(null)->get()->toArray();
$headers = ['Email', 'ID', 'Opens', 'Clicks'];
$csv = Writer::createFromPath($file, 'w+');
//insert the header
$csv->insertOne($headers);
//insert all the records
$csv->insertAll($records);
}
/**
* Campaign top 5 open location.
*
* @return \Illuminate\Database\Eloquent\Builder
*/
public function topLocations($number = 5)
{
$records = IpLocation::select('ip_locations.ip_address', 'ip_locations.country_code', 'ip_locations.country_name', 'ip_locations.region_name')
->addSelect(DB::raw('count(*) as aggregate'))
->join('open_logs', 'open_logs.ip_address', '=', 'ip_locations.ip_address')
->join('tracking_logs', 'open_logs.message_id', '=', 'tracking_logs.message_id')
->where('tracking_logs.campaign_id', '=', $this->id)
->join('campaigns', 'tracking_logs.campaign_id', '=', 'campaigns.id')
->where('campaigns.customer_id', '=', $this->customer->id)
->groupBy('ip_locations.ip_address', 'ip_locations.country_code', 'ip_locations.country_name', 'ip_locations.region_name')
->orderBy('aggregate', 'desc')
->take($number);
return $records;
}
public function topLocationsToCsv($file)
{
$records = $this->topLocations(null)->get()->toArray(); // null mean all
$headers = ['IP Address', 'Country Code', 'Country Name', 'Region Name', 'Count'];
$csv = Writer::createFromPath($file, 'w+');
//insert the header
$csv->insertOne($headers);
//insert all the records
$csv->insertAll($records);
}
/**
* Top 5 click locations (one row per clicked IP). Mirrors topLocations()
* but joins click_logs instead of open_logs.
*/
public function topClickLocations($number = 5)
{
return IpLocation::select('ip_locations.ip_address', 'ip_locations.country_code', 'ip_locations.country_name', 'ip_locations.region_name')
->addSelect(DB::raw('count(*) as aggregate'))
->join('click_logs', 'click_logs.ip_address', '=', 'ip_locations.ip_address')
->join('tracking_logs', 'click_logs.message_id', '=', 'tracking_logs.message_id')
->where('tracking_logs.campaign_id', '=', $this->id)
->join('campaigns', 'tracking_logs.campaign_id', '=', 'campaigns.id')
->where('campaigns.customer_id', '=', $this->customer->id)
->groupBy('ip_locations.ip_address', 'ip_locations.country_code', 'ip_locations.country_name', 'ip_locations.region_name')
->orderBy('aggregate', 'desc')
->take($number);
}
/**
* Campaign top 5 open countries.
*
* @return \Illuminate\Database\Eloquent\Builder
*/
public function topOpenCountries($number = 5)
{
$records = IpLocation::select('ip_locations.country_name', 'ip_locations.country_code')
->addSelect(DB::raw('count(*) as aggregate'))
->join('open_logs', 'open_logs.ip_address', '=', 'ip_locations.ip_address')
->join('tracking_logs', 'open_logs.message_id', '=', 'tracking_logs.message_id')
->where('tracking_logs.campaign_id', '=', $this->id)
->join('campaigns', 'tracking_logs.campaign_id', '=', 'campaigns.id')
->where('campaigns.customer_id', '=', $this->customer->id)
->groupBy('ip_locations.country_name', 'ip_locations.country_code')
->orderBy('aggregate', 'desc')
->take($number);
return $records;
}
public function topOpenCountriesToCsv($file)
{
$records = $this->topOpenCountries(null)->get()->toArray(); // null mean all
$headers = ['Country Name', 'Country Code', 'Open Count'];
$csv = Writer::createFromPath($file, 'w+');
//insert the header
$csv->insertOne($headers);
//insert all the records
$csv->insertAll($records);
}
/**
* Campaign top 5 click countries.
*
* @return \Illuminate\Database\Eloquent\Builder
*/
public function topClickCountries($number = 5)
{
$records = IpLocation::select('ip_locations.country_name', 'ip_locations.country_code')
->addSelect(DB::raw('count(*) as aggregate'))
->join('click_logs', 'click_logs.ip_address', '=', 'ip_locations.ip_address')
->join('tracking_logs', 'click_logs.message_id', '=', 'tracking_logs.message_id')
->join('campaigns', 'tracking_logs.campaign_id', '=', 'campaigns.id')
->where('tracking_logs.campaign_id', '=', $this->id)
->where('campaigns.customer_id', '=', $this->customer->id)
->groupBy('ip_locations.country_name', 'ip_locations.country_code')
->orderBy('aggregate', 'desc')
->take($number);
return $records;
}
public function topClickCountriesToCsv($file)
{
$records = $this->topClickCountries(null)->get()->toArray(); // null mean all
$headers = ['Country Name', 'Country Code', 'Click Count'];
$csv = Writer::createFromPath($file, 'w+');
//insert the header
$csv->insertOne($headers);
//insert all the records
$csv->insertAll($records);
}
/**
* Geocoded markers for the open-map page (one row per opened IP).
*/
public function openLocations()
{
return IpLocation::select(
'ip_locations.*',
'open_logs.created_at as open_at',
'subscribers.id as subscriber_id',
'subscribers.uid as subscriber_uid',
'subscribers.email as email',
'mail_lists.uid as list_uid'
)
->leftJoin('open_logs', 'open_logs.ip_address', '=', 'ip_locations.ip_address')
->leftJoin('tracking_logs', 'open_logs.message_id', '=', 'tracking_logs.message_id')
->leftJoin('subscribers', 'subscribers.id', '=', 'tracking_logs.subscriber_id')
->leftJoin('mail_lists', 'mail_lists.id', '=', 'subscribers.mail_list_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Geocoded markers for the click-map page (one row per clicked IP).
*/
public function clickLocations()
{
return IpLocation::select(
'ip_locations.*',
'click_logs.created_at as click_at',
'click_logs.url as url',
'subscribers.id as subscriber_id',
'subscribers.uid as subscriber_uid',
'subscribers.email as email',
'mail_lists.uid as list_uid'
)
->leftJoin('click_logs', 'click_logs.ip_address', '=', 'ip_locations.ip_address')
->leftJoin('tracking_logs', 'click_logs.message_id', '=', 'tracking_logs.message_id')
->leftJoin('subscribers', 'subscribers.id', '=', 'tracking_logs.subscriber_id')
->leftJoin('mail_lists', 'mail_lists.id', '=', 'subscribers.mail_list_id')
->where('tracking_logs.campaign_id', '=', $this->id);
}
/**
* Type of campaigns.
*
* @return object
*/
public static function types()
{
return [
'regular' => [
'icon' => 'attach_email',
],
'plain-text' => [
'icon' => 'wysiwyg',
],
'ab-test' => [
'icon' => 'science',
],
];
}
/**
* Copy new campaign.
*/
public function copy($name)
{
$copy = new self();
// Foreign key
$copy->customer_id = $this->customer_id;
$copy->default_mail_list_id = $this->default_mail_list_id;
$copy->tracking_domain_id = $this->tracking_domain_id;
// Overwrite attributes
$copy->name = $name;
$copy->status = self::STATUS_NEW;
// Overwrit date/time
$now = Carbon::now();
$copy->created_at = $now;
$copy->updated_at = $now;
// Other attributes to clone
$attributes = [
'type',
'subject',
'preheader',
'from_email',
'from_name',
'reply_to',
'sign_dkim',
'track_open',
'track_click',
'use_default_sending_server_from_email',
];
foreach ($attributes as $attribute) {
$copy[$attribute] = $this[$attribute];
}
// Save to DB
$copy->setConnection($this->customer->getDbConnection())->save();
// Lists segments
foreach ($this->listsSegments as $listSegment) {
$newListSegment = $copy->listsSegments()->make();
$newListSegment->mail_list_id = $listSegment->mail_list_id;
$newListSegment->segment_id = $listSegment->segment_id;
$newListSegment->created_at = $now;
$newListSegment->updated_at = $now;
$newListSegment->customer_id = $this->customer_id;
$newListSegment->setConnection($this->customer->getDbConnection())->save();
}
// Per-campaign sending server selection (empty = use full pool, copy as-is)
foreach ($this->sendingServersPivot as $row) {
$copy->sendingServersPivot()->create([
'sending_server_id' => $row->sending_server_id,
'customer_id' => $this->customer_id,
'fitness' => $row->fitness,
]);
}
// copy template
if (!is_null($this->template)) {
TemplateService::for($copy->email)->setTemplate($this->template, $copy->name);
$copy->email->updatePlainFromHtml();
$copy->unsetRelation('email');
$copy->updateLinks();
}
// refresh to update cache (otherwise, list-segment information will not be available yet)
$copy->refresh();
AppCache::for($copy)->refresh();
return $copy;
}
/**
* Send a test email for testing campaign.
*
* Credit accounting: a successful test send decrements 1 send_email credit
* (same key as a real campaign send). Test sends bypass SendPipeline so the
* usual credit gate doesn't fire — we check + consume here directly.
* Plans without a send_email grant or with zero remaining are blocked
* upstream of the SMTP send, before any side effect. Unlimited plans
* (Enterprise) pass canUse trivially and consume() is a no-op.
*/
public function sendTestEmail($email)
{
// @todo Find a better place for the following method, afterSave for example
$this->updateLinks();
$credits = app(\App\Services\Plans\Credits\CreditsService::class);
$sendKey = \App\Services\Plans\Credits\CreditKey::SEND_EMAIL;
$sub = $this->customer?->getCurrentActiveSubscription();
if ($sub === null) {
return [
'status' => 'error',
'message' => trans('messages.campaign.test_no_subscription'),
];
}
if (!$credits->canUse($sub, $sendKey, 1)) {
return [
'status' => 'error',
'message' => trans('messages.campaign.test_credit_exhausted'),
];
}
try {
// @todo: only send a test message when campaign sufficient information is available
// build a temporary subscriber oject used to pass through the sending methods
$subscriber = $this->mockSubscriber($email);
// Pick up an available sending server
// Throw exception in case no server available
$server = $this->pickSendingServer();
// build the message from campaign information
list($message, $msgId) = $this->prepareEmail($subscriber, $server);
// actually send
// @todo consider using queue here
$server->send($message);
// Consume only on a successful send so a failed SMTP attempt
// doesn't penalize the customer's credit balance. Unlimited plans
// short-circuit inside CreditsService::consume.
$credits->consume($sub, $sendKey, 1);
// Push the new live value to the DB cold snapshot immediately so
// the customer's bell / dashboard / quota panel reflect the test
// send without waiting for the user to click the "↻" refresh icon.
// Cheap (1-row UPDATE) and runs only on success.
$credits->syncSnapshotFromTracker($sub);
return [
'status' => 'success',
'message' => trans('messages.campaign.test_sent'),
];
} catch (\Exception $e) {
return [
'status' => 'error',
'message' => $e->getMessage(),
];
}
}
/**
* Get the delay time before sending.
*/
public function getDelayInSeconds()
{
$now = Carbon::now();
if ($now->gte($this->run_at)) {
return 0;
} else {
return (int)$this->run_at->diffInSeconds($now, $abs = true);
}
}
/**
* Re-send the campaign for sending.
*/
public function resend($filter = 'not_receive') // not_receive | not_open | not_click
{
// clean up failed log so that they will be included in resend
switch ($filter) {
case 'not_receive':
$this->cleanupFailedLog();
break;
case 'not_open':
$this->cleanupFailedLog();
$this->cleanupNotOpenLog();
break;
case 'not_click':
$this->cleanupFailedLog();
$this->cleanupNotClickLog();
break;
case 'force_resend':
$this->cleanupAllLogs();
break;
case 'resend_ab_test':
$this->cleanupAbTestForResend();
break;
default:
throw new \Exception("Unknown campaign RESEND type: ".$filter);
break;
}
// Force execute for full resend types
$force = in_array($filter, ['resend_ab_test', 'force_resend']);
// and queue again
$this->execute($force, ACM_QUEUE_TYPE_BATCH);
}
/**
* Re-send the campaign for sending.
*/
public function cleanupFailedLog()
{
// clean up failed log so that they will be included in resend
$recipients = $this->trackingLogs()->failed();
$recipients->delete();
}
public function cleanupNotOpenLog()
{
// clean up failed log so that they will be included in resend
$recipients = $this->trackingLogs()
->leftJoin('open_logs', 'tracking_logs.message_id', 'open_logs.message_id')
->whereNull('open_logs.id');
$recipients->delete();
}
public function cleanupNotClickLog()
{
// clean up failed log so that they will be included in resend
$recipients = $this->trackingLogs()
->leftJoin('click_logs', 'tracking_logs.message_id', 'click_logs.message_id')
->whereNull('click_logs.id');
$recipients->delete();
}
/**
* Clean up all AB test data and tracking logs for a full resend.
*/
public function cleanupAbTestForResend()
{
// Delete all derived logs that reference this campaign's tracking logs
$this->clickLogs()->delete();
$this->openLogs()->delete();
$this->feedbackLogs()->delete();
$this->bounceLogs()->delete();
$this->unsubscribeLogs()->delete();
// Delete all tracking logs
$this->trackingLogs()->delete();
// Reset AB test: delete assignments, clear winner
if ($this->hasAbTest()) {
$abTest = $this->abTest;
$abTest->assignments()->delete();
$abTest->update([
'winner_variant_id' => null,
'winner_sent_at' => null,
]);
}
// Reset campaign to fresh state
$this->status = self::STATUS_NEW;
$this->is_error = false;
$this->is_paused = false;
$this->last_error = null;
$this->save();
}
/**
* Clean up ALL tracking/delivery logs for a full force resend to every subscriber.
*/
public function cleanupAllLogs()
{
$this->clickLogs()->delete();
$this->openLogs()->delete();
$this->feedbackLogs()->delete();
$this->bounceLogs()->delete();
$this->unsubscribeLogs()->delete();
$this->trackingLogs()->delete();
// Reset campaign to fresh state
$this->status = self::STATUS_NEW;
$this->is_error = false;
$this->is_paused = false;
$this->last_error = null;
$this->save();
}
/**
* Get information from mail list.
*
* @param void
*/
public function getInfoFromMailList($list)
{
$this->from_name = !empty($this->from_name) ? $this->from_name : $list->from_name;
$this->from_email = !empty($this->from_email) ? $this->from_email : $list->from_email;
}
/**
* Get type select options.
*
* @return array
*/
public static function getTypeSelectOptions()
{
return [
['text' => trans('messages.'.Email::TYPE_REGULAR), 'value' => Email::TYPE_REGULAR],
['text' => trans('messages.'.Email::TYPE_PLAIN_TEXT), 'value' => Email::TYPE_PLAIN_TEXT],
];
}
/**
* The validation rules for automation trigger.
*
* @var array
*/
public function recipientsRules($params = [])
{
$rules = [
'lists_segments' => 'required',
];
if (isset($params['lists_segments'])) {
foreach ($params['lists_segments'] as $key => $param) {
$rules['lists_segments.'.$key.'.mail_list_uid'] = 'required';
}
}
return $rules;
}
/**
* Fill recipients by params.
*
* @var void
*/
public function fillRecipients($params = [])
{
if (isset($params['lists_segments'])) {
foreach ($params['lists_segments'] as $key => $param) {
$mail_list = null;
if (!empty($param['mail_list_uid'])) {
$mail_list = MailList::findByUid($param['mail_list_uid']);
// default mail list id
if (isset($param['is_default']) && $param['is_default'] == 'true') {
$this->default_mail_list_id = $mail_list->id;
}
}
if (!empty($param['segment_uids'])) {
foreach ($param['segment_uids'] as $segment_uid) {
$segment = Segment::findByUid($segment_uid);
$lists_segment = new CampaignsListsSegment();
$lists_segment->customer_id = $this->customer_id;
$lists_segment->campaign_id = $this->id;
if ($mail_list) {
$lists_segment->mail_list_id = $mail_list->id;
}
$lists_segment->segment_id = $segment->id;
$this->listsSegments->push($lists_segment);
}
} else {
$lists_segment = new CampaignsListsSegment();
$lists_segment->customer_id = $this->customer_id;
$lists_segment->campaign_id = $this->id;
if ($mail_list) {
$lists_segment->mail_list_id = $mail_list->id;
}
$this->listsSegments->push($lists_segment);
}
}
}
}
/**
* Save Recipients.
*
* @var void
*/
public function saveRecipients($params = [])
{
// Empty old data and refresh
$this->listsSegments()->delete();
$this->refresh();
// Fill params
$this->fillRecipients($params);
$lists_segments_groups = $this->getListsSegmentsGroups();
$data = [];
foreach ($lists_segments_groups as $lists_segments_group) {
if (!empty($lists_segments_group['segment_uids'])) {
foreach ($lists_segments_group['segment_uids'] as $segment_uid) {
$segment = Segment::findByUid($segment_uid);
$this->addListSegment($lists_segments_group['list'], $segment);
}
} else {
$this->addListSegment($lists_segments_group['list'], null);
}
}
// Save campaign with default list id
$campaign = Campaign::find($this->id);
$campaign->default_mail_list_id = $this->default_mail_list_id;
$campaign->setConnection($this->customer->getDbConnection())->save();
}
public function addListSegment($mail_list, $segment = null)
{
$lists_segment = new CampaignsListsSegment();
$lists_segment->customer_id = $this->customer_id;
$lists_segment->campaign_id = $this->id;
$lists_segment->mail_list_id = $mail_list->id;
if ($segment) {
$lists_segment->segment_id = $segment->id;
}
$lists_segment->save();
}
/**
* Display Recipients.
*
* @var array
*/
public function displayRecipients()
{
if (!$this->defaultMailList) {
return '';
}
$lines = [];
foreach ($this->getListsSegmentsGroups() as $lists_segments_group) {
if ($lists_segments_group['list']) {
$list_name = $lists_segments_group['list']->name;
$segment_names = [];
if (!empty($lists_segments_group['segment_uids'])) {
foreach ($lists_segments_group['segment_uids'] as $segment_uid) {
$segment = Segment::findByUid($segment_uid);
$segment_names[] = $segment->name;
}
}
if (empty($segment_names)) {
$lines[] = $list_name;
} else {
$lines[] = implode(': ', [$list_name, implode(', ', $segment_names)]);
}
}
}
return implode(' | ', $lines);
}
public function cacheManifest(): array
{
return [
// @note: SubscriberCount must come first as its value shall be used by the others
'ActiveSubscriberCount' => function () {
return $this->activeSubscribersCount(); // spepcial key that requires true update
},
'SubscriberCount' => function () {
return $this->subscribersCount(false); // spepcial key that requires true update
},
'DeliveredRate' => function () {
return $this->deliveredRate(true);
},
'DeliveredCount' => function () {
return $this->deliveredCount();
},
'FailedDeliveredRate' => function () {
return $this->failedRate(true);
},
'FailedDeliveredCount' => function () {
return $this->failedCount();
},
'NotDeliveredRate' => function () {
return $this->notDeliveredRate(true);
},
'NotDeliveredCount' => function () {
return $this->notDeliveredCount();
},
'PendingCount' => function () {
return $this->subscribersToSend()->count();
},
'SkippedCount' => function () {
return $this->subscribersSkipped()->count();
},
'ClickCount' => function () {
return $this->clickCount();
},
'ClickedRate' => function () {
return $this->clickRate();
},
'UniqOpenRate' => function () {
return $this->openRate();
},
'UniqOpenCount' => function () {
return $this->openUniqCount();
},
'NotOpenRate' => function () {
return $this->notOpenRate();
},
'NotOpenCount' => function () {
return $this->notOpenCount();
},
'BounceCount' => function () {
return $this->bounceCount();
},
'UnsubscribeCount' => function () {
return $this->unsubscribeCount();
},
'FeedbackCount' => function () {
return $this->feedbackCount();
},
'UniqueClickCount' => function () {
return $this->uniqueClickCount();
},
'BounceRate' => function () {
return $this->bounceRate();
},
'UnsubscribeRate' => function () {
return $this->unsubscribeRate();
},
'FeedbackRate' => function () {
return $this->feedbackRate();
},
'AbuseFeedbackCount' => function () {
return $this->abuseFeedbackCount();
},
// Section-shaped: every block on the overview's "top X" cards.
// One atomic refresh, one read in the controller. Eloquent
// collections serialize fine; we stash Carbon for the
// last-event timestamps to avoid hauling whole log rows.
'OverviewTopBlocks' => [
'compute' => function () {
return [
'top_links' => $this->getTopLinks(5)->get(),
'top_open_countries' => $this->topOpenCountries(7)->get(),
'top_click_countries' => $this->topClickCountries(7)->get(),
'top_active_subscribers' => $this->getTopActiveSubscribers(5)->get(),
'top_locations' => $this->topLocations(5)->get(),
'last_open_at' => optional($this->lastOpen())->created_at,
'last_click_at' => optional($this->lastClick())->created_at,
];
},
'ttl' => 3600,
],
];
}
/**
* Oldest write timestamp among the big-table cache entries that back the
* campaign overview page. Null = nothing computed yet ("never").
*/
public function getOverviewComputedAt(): ?Carbon
{
$scope = AppCache::for($this);
$timestamps = array_filter([
$scope->lastUpdatedAt('SubscriberCount'),
$scope->lastUpdatedAt('DeliveredCount'),
$scope->lastUpdatedAt('UniqOpenCount'),
$scope->lastUpdatedAt('ClickCount'),
$scope->lastUpdatedAt('OverviewTopBlocks'),
]);
return $timestamps ? min($timestamps) : null;
}
/**
* Count subscribers.
*
* @return int
*/
public function subscribersCount($cache = false)
{
if ($cache) {
return AppCache::for($this)->read('SubscriberCount', 0);
}
return $this->subscribers([])->count();
}
/**
* Count subscribers.
*
* @return int
*/
public function activeSubscribersCount()
{
// return distinctCount($this->subscribers([])->where('subscribers.status', Subscriber::STATUS_SUBSCRIBED), 'subscribers.email');
return $this->subscribers([])->where('subscribers.status', Subscriber::STATUS_SUBSCRIBED)->count();
}
/**
* Count unique open by hour.
*
* @return number
*/
public function openUniqHours($start = null, $end = null)
{
$query = $this->openLogs()->select('open_logs.created_at');
$currentTimezone = $this->customer->getTimezone();
if (isset($start)) {
$query = $query->where('open_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('open_logs.created_at', '<=', $end);
}
return $query->orderBy('open_logs.created_at', 'asc')->get()->groupBy(function ($date) use ($currentTimezone) {
return $date->created_at->timezone($currentTimezone)->format('H'); // grouping by hours
});
}
/**
* Count click group by hour.
*
* @return number
*/
public function clickHours($start = null, $end = null)
{
$currentTimezone = $this->customer->getTimezone();
$query = $this->clickLogs()->select('click_logs.created_at', 'tracking_logs.subscriber_id');
if (isset($start)) {
$query = $query->where('click_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('click_logs.created_at', '<=', $end);
}
return $query->orderBy('click_logs.created_at', 'asc')->get()->groupBy(function ($date) use ($currentTimezone) {
return $date->created_at->timezone($currentTimezone)->format('H'); // grouping by hours
});
}
public function fileInfo($filePath)
{
$name = $filePath['filename'];
$extension = $filePath['extension'];
return $name.'.'.$extension;
}
public function fillAttributes($params)
{
$this->fill($params);
// Tacking domain
if (isset($params['custom_tracking_domain']) && $params['custom_tracking_domain'] && isset($params['tracking_domain_uid'])) {
$tracking_domain = \App\Model\TrackingDomain::findByUid($params['tracking_domain_uid']);
if ($tracking_domain) {
$this->tracking_domain_id = $tracking_domain->id;
} else {
$this->tracking_domain_id = null;
}
} else {
$this->tracking_domain_id = null;
}
}
/**
* Generate SpamScore.
*/
public function score()
{
// raw output
$test = $this->execSpamc();
// Get scores / thresholds
preg_match('/\s*(?<score>[0-9\.\/]+)\s*/', $test, $score);
if (!array_key_exists('score', $score)) {
throw new \Exception('Cannot get SpamScore: '.$test);
}
$score = $score['score'];
list($current, $threshold) = preg_split('/\//', $score);
$passed = ($current <= $threshold) ? true : false;
// get the details
$json = [];
$firstMatch = false;
foreach (preg_split("/((\r?\n)|(\r\n?))/", $test) as $line) {
preg_match('/^\s*(?<score>[\-0-9\.]+)\s+(?<rule>[\w]+)\s+(?<desc>.*)/', $line, $result);
if (array_key_exists('score', $result) && array_key_exists('rule', $result) && array_key_exists('desc', $result)) {
$firstMatch = true;
$json[] = [
'score' => $result['score'],
'rule' => $result['rule'],
'desc' => $result['desc'],
'status' => ($result['score'] > 0.0) ? 'failed' : (($result['score'] == 0.0) ? 'neutral' : 'passed'),
];
} elseif ($firstMatch) {
$lastRecord = end($json);
$lastRecord['desc'] .= ' '.trim($line);
// replace last record
$json[sizeof($json) - 1] = $lastRecord;
}
}
return [
'result' => $passed,
'score' => $score,
'details' => $json,
];
}
/**
* Generate SpamScore.
*/
private function execSpamc()
{
$message = $this->getSampleMessage()->toString();
// Execute SPAMC
$desc = [
0 => array('pipe', 'r'), // 0 is STDIN for process
1 => array('pipe', 'w'), // 1 is STDOUT for process
2 => array('pipe', 'w'), // 2 is STDERR for process
];
// command to invoke markup engine
$cmd = Setting::get('spamassassin.command');
if (is_null($cmd)) {
$cmd = 'spamc -R'; // default value
}
// spawn the process
$p = proc_open($cmd, $desc, $pipes);
// send the wiki content as input to the markup engine
// and then close the input pipe so the engine knows
// not to expect more input and can start processing
fwrite($pipes[0], $message);
fclose($pipes[0]);
// read the output from the engine
$stdout = stream_get_contents($pipes[1]);
$stderr = stream_get_contents($pipes[2]);
// all done! Clean up
fclose($pipes[1]);
fclose($pipes[2]);
proc_close($p);
if (!empty($err) || empty($stdout)) {
throw new \Exception('Error: cannot get SpamScore: '.$stderr);
}
return $stdout;
}
/**
* Get sample message.
*/
public function getSampleMessage()
{
// Build a valid message with a fake contact
// build a temporary subscriber oject used to pass through the sending methods
$subscriber = $this->mockSubscriber('[email protected]');
// Throw exception in case no server available
$server = $this->pickSendingServer();
// build the message from campaign information
list($message, $msgId) = $this->prepareEmail($subscriber, $server);
return $message;
}
/**
* Copy new template.
*/
public function generateTrackingLogCsv($logtype, $progressCallback = null)
{
$tmpTableName = 'log_'.$this->uid.'_'.md5(rand());
$filePath = storage_path(join_paths('app', $tmpTableName.'.csv'));
DB::statement('DROP TABLE IF EXISTS '.table($tmpTableName));
if ($logtype == 'open_logs') {
DB::statement('CREATE TEMPORARY TABLE '.table($tmpTableName).' AS SELECT
c.name as campaign_name,
s.email as subscriber_email,
l.status as delivery_status,
l.created_at as sent_at,
o.created_at AS open_at,
o.user_agent AS device,
ip.ip_address,
ip.country_code,
ip.country_name,
ip.region_code,
ip.region_name,
ip.city,
ip.zipcode,
ip.latitude,
ip.longitude,
ip.metro_code
FROM '.table('tracking_logs').' l
JOIN '.table('campaigns').' c ON l.campaign_id = c.id
JOIN '.table('subscribers').' s ON l.subscriber_id = s.id
JOIN '.table('open_logs').' o ON o.message_id = l.message_id
LEFT JOIN '.table('ip_locations').' ip ON ip.ip_address = o.ip_address
WHERE c.id = '.$this->id. ' ORDER BY c.name, l.created_at');
$headers = [
'campaign_name',
'subscriber_email',
'delivery_status',
'sent_at',
'open_at',
'device',
'ip_address',
'country_code',
'country_name',
'region_code',
'region_name',
'city',
'zipcode',
'latitude',
'longitude',
'metro_code',
];
} elseif ($logtype == 'click_logs') {
DB::statement('CREATE TEMPORARY TABLE '.table($tmpTableName).' AS SELECT
c.name as campaign_name,
s.email as subscriber_email,
l.status as delivery_status,
l.created_at as sent_at,
ck.created_at AS click_at,
ck.url,
ck.user_agent AS device,
ip.ip_address,
ip.country_code,
ip.country_name,
ip.region_code,
ip.region_name,
ip.city,
ip.zipcode,
ip.latitude,
ip.longitude,
ip.metro_code
FROM '.table('tracking_logs').' l
JOIN '.table('campaigns').' c ON l.campaign_id = c.id
JOIN '.table('subscribers').' s ON l.subscriber_id = s.id
JOIN '.table('click_logs').' ck ON ck.message_id = l.message_id
LEFT JOIN '.table('ip_locations').' ip ON ip.ip_address = ck.ip_address
WHERE c.id = '.$this->id. ' ORDER BY c.name, l.created_at');
$headers = [
'campaign_name',
'subscriber_email',
'delivery_status',
'sent_at',
'click_at',
'url',
'device',
'ip_address',
'country_code',
'country_name',
'region_code',
'region_name',
'city',
'zipcode',
'latitude',
'longitude',
'metro_code',
];
} elseif ($logtype == 'unsubscribe_logs') {
DB::statement('CREATE TEMPORARY TABLE '.table($tmpTableName).' AS SELECT
c.name as campaign_name,
s.email as subscriber_email,
l.status as delivery_status,
l.created_at as sent_at,
u.created_at AS unsubscribe_at,
u.user_agent AS device,
ip.ip_address,
ip.country_code,
ip.country_name,
ip.region_code,
ip.region_name,
ip.city,
ip.zipcode,
ip.latitude,
ip.longitude,
ip.metro_code
FROM '.table('tracking_logs').' l
JOIN '.table('campaigns').' c ON l.campaign_id = c.id
JOIN '.table('subscribers').' s ON l.subscriber_id = s.id
JOIN '.table('unsubscribe_logs').' u ON u.message_id = l.message_id
LEFT JOIN '.table('ip_locations').' ip ON ip.ip_address = u.ip_address
WHERE c.id = '.$this->id. ' ORDER BY c.name, l.created_at');
$headers = [
'campaign_name',
'subscriber_email',
'delivery_status',
'sent_at',
'unsubscribe_at',
'device',
'ip_address',
'country_code',
'country_name',
'region_code',
'region_name',
'city',
'zipcode',
'latitude',
'longitude',
'metro_code',
];
} elseif ($logtype == 'feedback_logs') {
DB::statement('CREATE TEMPORARY TABLE '.table($tmpTableName).' AS SELECT
c.name AS campaign_name,
s.email AS subscriber_email,
l.status AS delivery_status,
l.created_at AS sent_at,
f.created_at AS feedback_at,
f.feedback_type AS feedback_type,
f.raw_feedback_content AS feedback_content
FROM '.table('tracking_logs').' l
JOIN '.table('campaigns').' c ON l.campaign_id = c.id
JOIN '.table('subscribers').' s ON l.subscriber_id = s.id
JOIN '.table('feedback_logs').' f ON f.message_id = l.message_id
WHERE c.id = '.$this->id. ' ORDER BY c.name, l.created_at');
$headers = [
'campaign_name',
'subscriber_email',
'delivery_status',
'sent_at',
'feedback_at',
'feedback_type',
'feedback_content',
];
} elseif ($logtype == 'bounce_logs') {
DB::statement('CREATE TEMPORARY TABLE '.table($tmpTableName).' AS SELECT
c.name AS campaign_name,
s.email AS subscriber_email,
l.status AS delivery_status,
l.created_at AS sent_at,
b.created_at AS bounce_at,
b.bounce_type AS bounce_type,
b.raw AS bounce_content
FROM '.table('tracking_logs').' l
JOIN '.table('campaigns').' c ON l.campaign_id = c.id
JOIN '.table('subscribers').' s ON l.subscriber_id = s.id
JOIN '.table('bounce_logs').' b ON b.message_id = l.message_id
WHERE c.id = '.$this->id. ' ORDER BY c.name, l.created_at');
$headers = [
'campaign_name',
'subscriber_email',
'delivery_status',
'sent_at',
'bounce_at',
'bounce_type',
'bounce_content',
];
} elseif ($logtype == 'tracking_logs') {
DB::statement('CREATE TEMPORARY TABLE '.table($tmpTableName).' AS SELECT
c.name AS campaign_name,
s.email AS subscriber_email,
l.status AS delivery_status,
l.created_at AS sent_at
FROM '.table('tracking_logs').' l
JOIN '.table('campaigns').' c ON l.campaign_id = c.id
JOIN '.table('subscribers').' s ON l.subscriber_id = s.id
WHERE c.id = '.$this->id. ' ORDER BY c.name, l.created_at');
$headers = [
'campaign_name',
'subscriber_email',
'delivery_status',
'sent_at',
];
} else {
throw new \Exception('Unknown export type: '.$logtype);
}
$total = DB::table($tmpTableName)->count();
$limit = 1000;
$pages = ceil($total / $limit);
// insert header
$csv = Writer::createFromPath($filePath, 'w+');
$csv->insertOne($headers);
for ($i = 0; $i < $pages; $i += 1) {
$items = DB::table($tmpTableName)->select('*')
->limit($limit)
->offset($i * $limit)
->get()
->map(function ($r) {
return (array) $r;
})
->toArray();
$csv->insertAll($items);
// callback progress
if (!is_null($progressCallback)) {
$percentage = ($i + 1) / $pages;
$progressCallback($percentage, $filePath);
}
}
// callback progress
if (!is_null($progressCallback)) {
$progressCallback($percentage = 100, $filePath);
}
return $filePath;
}
public function arrayRandFixed($array, $count)
{
$result = array_rand($array, $count);
if (is_array($result)) {
return $result;
} else {
return [$result];
}
}
public function getDeliveryReport()
{
/****
* Important: there are 4 possible values of `delivery_stats`
* + sent
* + failed
* + new
* + skipped
*
* Assumption
* + subscribers 1:1 email_verification
* + [campaign, subscriber] 1:1 tracking_logs
* */
$query = $this->subscribers()->leftJoinSub(
// Why joinSub? Notice that this->subscribers() does not have campaign_id constraint!
// while this->trackingLogs() does have
$this->trackingLogs(),
'tracking_logs',
function ($join) {
$join->on('tracking_logs.subscriber_id', 'subscribers.id');
}
)->leftJoin('bounce_logs', 'tracking_logs.id', 'bounce_logs.tracking_log_id')
->leftJoin('feedback_logs', 'tracking_logs.id', 'feedback_logs.tracking_log_id')
->leftJoin('unsubscribe_logs', 'tracking_logs.message_id', 'unsubscribe_logs.message_id');
$query->select(DB::raw(strtr("
CASE
WHEN `%bounce_logs`.`id` IS NOT NULL THEN '%bounced'
WHEN `%feedback_logs`.`id` IS NOT NULL THEN '%feedback'
WHEN `%unsubscribe_logs`.`id` IS NOT NULL THEN '%unsubscribed'
ELSE
CASE `%tracking_logs`.`status`
WHEN '%sent' THEN '%sent'
WHEN '%failed' THEN '%failed'
ELSE
CASE `%subscribers`.`status`
WHEN '%subscribed' THEN
CASE COALESCE(`%subscribers`.`verification_status`, '-1')
WHEN '-1' THEN '%new'
WHEN '%deliverable' THEN '%new'
ELSE '%skipped'
END
ELSE '%skipped'
END
END
END
AS delivery_status
", [
'%sent' => self::DELIVERY_STATUS_SENT,
'%failed' => self::DELIVERY_STATUS_FAILED,
'%new' => self::DELIVERY_STATUS_NEW,
'%skipped' => self::DELIVERY_STATUS_SKIPPED,
'%bounced' => self::DELIVERY_STATUS_BOUNCED,
'%feedback' => self::DELIVERY_STATUS_FEEDBACK,
'%subscribed' => Subscriber::STATUS_SUBSCRIBED,
'%unsubscribed' => Subscriber::STATUS_UNSUBSCRIBED,
'%deliverable' => Subscriber::VERIFICATION_STATUS_DELIVERABLE,
'%tracking_logs' => table('tracking_logs'),
'%subscribers' => table('subscribers'),
'%bounce_logs' => table('bounce_logs'),
'%feedback_logs' => table('feedback_logs'),
'%unsubscribe_logs' => table('unsubscribe_logs')
])));
return $query;
}
public function getDeliveryReportSummary()
{
$query = DB::query()->fromSub($this->getDeliveryReport(), 'campaign_subscribers')->groupBy('delivery_status')->select(DB::raw('delivery_status, COUNT(1)'));
return $query;
}
public function generateWebViewerPreviewUrl($subscriber)
{
return route('webViewerPreviewUrl', [
'campaign_uid' => $this->uid,
'subscriber_id' => $subscriber->id,
]);
}
public function isArchived()
{
return static::ARCHIVE_STATUS_YES == $this->archive_status;
}
public function prepare()
{
// Available sending servers — resolved through Customer (own) or Plan
// (admin pool); MailList is no longer a server wrapper.
$servers = $this->customer->getSendingServerPool();
// No sending server ready for delivery
if (empty($servers)) {
throw new Exception('No sending server available');
}
// @important: make sure this function is called twice at the same time.
// Otherwise, concurrency issue might happen: "Campaign stopped. Timeout getting lock #Lockable for: /home/app/public_html/storage/locks/sending-server-sns-6512d..."
foreach ($servers as $serverId => $fitness) {
$server = SendingServer::find($serverId);
if ($this->useSendingServerFromEmailAddress()) {
$fromEmailAddress = $server->default_from_email;
} else {
$fromEmailAddress = $this->from_email;
}
$this->logger()->info('Setting up sending server before send: '.$server->uid);
if (!config('custom.dryrun')) {
$server->driver()->setupBeforeSend($fromEmailAddress);
}
$server->getRateLimitTracker()->cleanup('24 hours');
// enable warmup server if needed
$server->enableWarmupIfNeeded();
}
// Reset max_execution_time so that command can run for a long time without being terminated
self::resetMaxExecutionTime();
// Clear any HTML cache
if ($this->hasAbTest()) {
foreach ($this->abTest->variants()->with('email')->get() as $variant) {
if ($variant->email) {
$variant->email->clearTemplateCache();
}
}
} else {
$this->clearTemplateCache();
}
// Last chance for campaign to setup
// @todo Find a better place for the following method, afterSave for example
$this->updateLinks();
$subscription = $this->customer->getCurrentActiveSubscription();
if ($subscription !== null) {
$rateTracker = app(\App\Services\Plans\RateLimits\RateLimitsService::class)
->trackerFor($subscription, \App\Services\Plans\RateLimits\RateLimitKey::SEND_EMAIL_RATE);
// null = plan does not configure the rate limit, nothing to prune.
if ($rateTracker !== null && method_exists($rateTracker, 'cleanup')) {
$rateTracker->cleanup('24 hours');
}
}
}
public function getServersPool()
{
if (!is_null($this->serversPool)) {
return $this->serversPool;
}
// ─── Case B short-circuit ─────────────────────────────────────────
// When the customer's plan doesn't grant HAS_OWN_SENDING_SERVER the
// pool is admin-managed via the plan. Any pivot rows on this campaign
// would be leftovers from a prior tier (customer used to be Case A
// then downgraded). Treat them as forfeited — Case B campaigns ALWAYS
// resolve through the plan pool so a stale narrow never leaks
// forbidden servers into a downgraded send.
if (!$this->canPickSendingServers()) {
return $this->serversPool = $this->buildDefaultPool();
}
// ─── Nhánh B/C: campaign has explicit narrow selection ───
$pivot = $this->sendingServersPivot()->with('sendingServer')->get();
if ($pivot->isNotEmpty()) {
$pool = new RouletteWheel();
foreach ($pivot as $row) {
$server = $row->sendingServer;
if (!$server || $server->status !== SendingServer::STATUS_ACTIVE) {
continue;
}
$pool->add($server, (int) $row->fitness);
}
// Nhánh C: all selected servers inactive → fail loud.
// Respect narrow intent — never silently fall back to the full pool.
if ($pool->count() === 0) {
throw new Exception(sprintf(
'Campaign %s: all per-campaign sending servers are inactive',
$this->id
));
}
return $this->serversPool = $pool;
}
// ─── Nhánh A: pivot empty → default "use all from pool" ───
return $this->serversPool = $this->buildDefaultPool();
}
/**
* Build the default-broadcast RouletteWheel from the customer's pool.
* Shared between nhánh A (pivot empty, Case A campaign) and the Case B
* short-circuit above. Throws when the pool is empty — that's a
* misconfiguration the admin/customer must fix (no own servers AND no
* plan servers granted).
*/
private function buildDefaultPool(): RouletteWheel
{
$serversAndWeights = $this->customer->getSendingServerPool();
if (empty($serversAndWeights)) {
throw new Exception('No sending server available');
}
$pool = new RouletteWheel();
foreach ($serversAndWeights as $serverId => $weight) {
$server = SendingServer::find($serverId);
$pool->add($server, $weight);
}
return $pool;
}
/**
* Whether the customer is allowed to narrow the campaign's server pool.
* True when their active plan grants HAS_OWN_SENDING_SERVER — the same
* flag that lets MailList::getSendingServers() use customer-owned servers
* instead of plan-granted ones. When false, the pool is fully determined
* by the plan and the campaign UI shows it read-only.
*/
public function canPickSendingServers(): bool
{
return EntitlementGate::allows($this->customer, EntitlementKey::HAS_OWN_SENDING_SERVER);
}
/**
* IDs of sending servers eligible for this campaign — the union returned
* by Customer::getSendingServerPool() at edit time. Used to validate that
* a narrow selection only contains servers the customer is entitled to.
*
* @return array<int,int>
*/
public function availableSendingServerIds(): array
{
if (!$this->customer) {
return [];
}
$ids = array_keys($this->customer->getSendingServerPool());
return array_map('intval', $ids);
}
/**
* Available sending servers (full models) for picker UI, in the eligible
* pool order. Each row exposes id, name, type via standard accessors.
*
* @return \Illuminate\Support\Collection<int,\App\Model\SendingServer>
*/
public function availableSendingServers()
{
$ids = $this->availableSendingServerIds();
if (empty($ids)) {
return collect();
}
return SendingServer::whereIn('id', $ids)->orderBy('name')->get();
}
/**
* Replace the campaign's narrow selection with $rows. Empty array clears
* the pivot — campaign reverts to the default "use all from pool" path.
*
* Case B customers (plan doesn't grant HAS_OWN_SENDING_SERVER) NEVER hold
* pivot rows: their pool is admin-managed via the plan. Any incoming
* $rows for such customers are dropped on the floor — defense in depth
* against tampered POST bodies, mis-wired callers, or stale state from a
* downgraded customer who used to be Case A. Existing pivot rows (from
* before the downgrade) get cleared by the unconditional DELETE below.
*
* @param array<int,array{sending_server_id:int|string,fitness?:int|string}> $rows
*/
public function syncSendingServers(array $rows): void
{
if (!$this->canPickSendingServers()) {
$rows = [];
}
$this->sendingServersPivot()->delete();
// Bust memoized pool so the next send picks up the new selection.
$this->serversPool = null;
if (empty($rows)) {
return;
}
foreach ($rows as $row) {
$serverId = (int) ($row['sending_server_id'] ?? 0);
if ($serverId <= 0) {
continue;
}
$this->sendingServersPivot()->create([
'sending_server_id' => $serverId,
'customer_id' => $this->customer_id,
'fitness' => max(1, (int) ($row['fitness'] ?? 100)),
]);
}
}
// @important
// This method retrieve subscribers that are not associated with the campaign in the tracking_logs table, i.e. not sent
// However, how about pending SendMessage job? i.e. executing this method again in parallel will cause duplicate SendMessage jobs
// intended for the same subscribers.
//
// Luckily, this method is called by the LoadCampaign job in a batch only!
// It means: + Only one LoadCampaign job is alive at a given time
// + All the SendMessage jobs must be done before dispatching the next LoadCampaign job (the "then" event of Batch)
// + It guarantees that this method is not called the second time before all SendMessage jobs are already done.
public function subscribersToSend()
{
// Notice that if an email already exists in tracking_logs but with status 'duplicate' (skipped) => then a new message to the same email will be allowed
// Retrieve subscribers to send!
$query = $this->subscribers([])
->whereRaw(sprintf(table('subscribers').".email NOT IN (SELECT email FROM %s t JOIN %s s ON t.subscriber_id = s.id WHERE t.campaign_id = %s AND t.status != '%s')", table('tracking_logs'), table('subscribers'), $this->id, TrackingLog::STATUS_DUPLICATE))
->subscribed()
->byVerificationStatus($this->getDeliveryStatuses());
return $query;
}
/**
* (C) Subscribers already sent — have a tracking_log entry for this campaign
* (regardless of tracking_log status: sent, error, etc.)
*/
public function subscribersSent()
{
return $this->subscribers([])
->whereRaw(sprintf(
table('subscribers') . ".email IN (SELECT email FROM %s t JOIN %s s ON t.subscriber_id = s.id WHERE t.campaign_id = %s AND t.status != '%s')",
table('tracking_logs'),
table('subscribers'),
$this->id,
TrackingLog::STATUS_DUPLICATE
));
}
/**
* (D) Subscribers skipped — in subscribers([]) but NOT in subscribersToSend() and NOT in subscribersSent().
* Excluded due to status (unsubscribed/blacklisted) or verification status not meeting delivery criteria.
*/
public function subscribersSkipped()
{
// Start from all subscribers (A), exclude those already sent (C)
$query = $this->subscribers([])
->whereRaw(sprintf(
table('subscribers') . ".email NOT IN (SELECT email FROM %s t JOIN %s s ON t.subscriber_id = s.id WHERE t.campaign_id = %s AND t.status != '%s')",
table('tracking_logs'),
table('subscribers'),
$this->id,
TrackingLog::STATUS_DUPLICATE
));
// Exclude those eligible to send (B) — NOT (subscribed AND allowed verification)
$statuses = $this->getDeliveryStatuses();
$query->where(function ($q) use ($statuses) {
$q->where('subscribers.status', '!=', Subscriber::STATUS_SUBSCRIBED);
if (in_array(Subscriber::VERIFICATION_STATUS_UNVERIFIED, $statuses)) {
$q->orWhere(function ($q2) use ($statuses) {
$q2->whereNotNull('subscribers.verification_status')
->whereNotIn('subscribers.verification_status', $statuses);
});
} else {
$q->orWhereNotIn('subscribers.verification_status', $statuses);
}
});
return $query;
}
/**
* Count subscribers remaining to send.
* For A/B campaigns: only count subscribers that have an assignment —
* unassigned remainder (winner phase) must not trigger a ContinueCampaign loop.
*/
public function countSubscribersToSend(): int
{
if ($this->hasAbTest()) {
$abTestId = $this->abTest()->value('id');
return $this->subscribersToSend()
->join('ab_test_assignments as ata_count', function ($join) use ($abTestId) {
$join->on('ata_count.subscriber_id', '=', 'subscribers.id')
->where('ata_count.ab_test_id', '=', $abTestId);
})
->count();
}
return $this->subscribersToSend()->count();
}
// This is a better version without "NOT IN" which might cause performance issue
// However, we need to do more testing before switching
public function subscribersToSend2()
{
// Notice that if an email already exists in tracking_logs but with status 'duplicate' (skipped) => then a new message to the same email will be allowed
// Retrieve subscribers to send!
$query = $this->subscribers([])
->whereNotExists(function ($query) {
$query->selectRaw(1)
->from('tracking_logs as t')
->join('subscribers as s', 't.subscriber_id', '=', 's.id')
->whereColumn('s.email', 'subscribers.email') // correlate by email
->where('t.campaign_id', $this->id)
->where('t.status', '!=', TrackingLog::STATUS_DUPLICATE);
})
->subscribed()
->byVerificationStatus($this->getDeliveryStatuses());
return $query;
}
/**
* Subscribers.
*
* @return collect
*/
public function subscribers($params = [])
{
// Get subscriber from mailist and segment
$listsAndSegments = [];
foreach ($this->listsSegments as $lists_segment) {
if (!empty($lists_segment->segment_id)) {
$listsAndSegments[] = $lists_segment->segment;
} else {
$listsAndSegments[] = $lists_segment->mailList;
}
}
$query = Subscriber::getByListsAndSegments($this->customer->getDbConnection(), ...$listsAndSegments);
// Filters
$filters = isset($params['filters']) ? $params['filters'] : null;
if ((isset($filters) && (isset($filters['open']) || isset($filters['click']) || isset($filters['tracking_status'])))
) {
$query = $query->leftJoin('tracking_logs', 'tracking_logs.subscriber_id', '=', 'subscribers.id');
$query = $query->whereNotNull('tracking_logs.id');
$query = $query->where('tracking_logs.campaign_id', '=', $this->id);
}
if (isset($filters)) {
if (isset($filters['open'])) {
$equal = ($filters['open'] == 'opened') ? 'whereNotNull' : 'whereNull';
$query = $query->leftJoin('open_logs', 'tracking_logs.message_id', '=', 'open_logs.message_id')
->$equal('open_logs.id');
}
if (isset($filters['click'])) {
$equal = ($filters['click'] == 'clicked') ? 'whereNotNull' : 'whereNull';
$query = $query->leftJoin('click_logs', 'tracking_logs.message_id', '=', 'click_logs.message_id')
->$equal('click_logs.id');
}
if (isset($filters['tracking_status'])) {
$val = ($filters['tracking_status'] == 'not_sent') ? null : $filters['tracking_status'];
$query = $query->where('tracking_logs.status', '=', $val);
}
}
// keyword
if (isset($params['keyword']) && !empty(trim($params['keyword']))) {
foreach (explode(' ', trim($params['keyword'])) as $keyword) {
$query = $query->where(function ($q) use ($keyword) {
$q->orwhere('subscribers.email', 'like', '%'.$keyword.'%');
});
}
}
return $query;
}
/**
* Pick up a delivery server for the campaign.
*
* @return mixed
*/
public function pickSendingServer()
{
return $this->customer->pickSendingServer();
}
public function doSendTestEmail($email)
{
$validator = \Validator::make(['email' => $email], [
'email' => 'required|email',
]);
// redirect if fails
if ($validator->fails()) {
return $validator;
}
// validate service
$validator->after(function ($validator) use ($email) {
try {
$result = $this->sendTestEmail($email);
if ($result['status'] == 'error') {
$validator->errors()->add('email', 'Can not send test email. Error: ' . $result['message']);
}
} catch (\Exception $e) {
$validator->errors()->add('email', 'Can not send test email. Error: ' . $e->getMessage());
}
});
return $validator;
}
public function newWebhook()
{
$webhook = new \App\Model\CampaignWebhook();
$webhook->campaign_id = $this->id;
$webhook->customer_id = $this->customer_id;
return $webhook;
}
public function queueOpenCallbacks($log)
{
$callbacks = $this->campaignWebhooks()->open()->get();
foreach ($callbacks as $callback) {
ExecuteCampaignCallback::dispatch($callback, $log);
}
}
public function queueClickCallbacks($log)
{
$callbacks = $this->campaignWebhooks()
->click()
// ->join('campaign_links', 'campaign_webhooks.campaign_link_id', 'campaign_links.id')
// ->where('url', $log->url)
->get();
foreach ($callbacks as $callback) {
ExecuteCampaignCallback::dispatch($callback, $log);
}
}
public function queueUnsubscribeCallbacks($log)
{
$callbacks = $this->campaignWebhooks()->unsubscribe()->get();
foreach ($callbacks as $callback) {
ExecuteCampaignCallback::dispatch($callback, $log);
}
}
public function openWebhooks()
{
return $this->campaignWebhooks()->open();
}
public function isStageExcluded(string $name): bool
{
switch ($name) {
case InjectTrackingPixel::class:
if ($this->track_open) {
return false;
} else {
return true;
}
break;
default:
// do not exclude by default
return false;
break;
}
}
public function stopOnError()
{
return $this->skip_failed_message == false;
}
public static function newDefault()
{
$campaign = new self([
'track_open' => true,
'track_click' => true,
'sign_dkim' => true,
]);
$campaign->type = Email::TYPE_REGULAR;
$campaign->name = trans('messages.untitled');
$campaign->status = Campaign::STATUS_NEW;
$campaign->skip_failed_message = Setting::isYes('email.default.skip_failed_message');
$campaign->fillDeliveryStatuses($campaign->getDefaultDeliveryStatuses());
return $campaign;
}
public function saveFromArray($params)
{
// custom campaign content
$content = null;
if (isset($params['html'])) {
$content = $params['html'];
unset($params['html']);
}
if (isset($params['track_open'])) {
$params['track_open'] = filter_var($params['track_open'], FILTER_VALIDATE_BOOLEAN);
}
if (isset($params['track_click'])) {
$params['track_click'] = filter_var($params['track_click'], FILTER_VALIDATE_BOOLEAN);
}
if (isset($params['sign_dkim'])) {
$params['sign_dkim'] = filter_var($params['sign_dkim'], FILTER_VALIDATE_BOOLEAN);
}
$this->fill($params);
// type
if (isset($params['type'])) {
$this->type = $params['type'];
}
// save
$this->save();
// Mail list
if (isset($params['mail_list_uid'])) {
$mailList = \App\Model\MailList::findByUid($params['mail_list_uid']);
// Insert Data
$this->addListSegment($mailList, null);
$this->default_mail_list_id = $mailList->id;
$this->save();
}
}
public function makeSampleData($params = [])
{
$this->cleanSampleData();
$default = [
'unconfirmed' => 0.08,
'delivery_failed' => 0.1, // against all deivered emails
'open' => 0.45, // against delivery_success
'click' => 0.85, // against open
'unsubscribe' => 0.2, // against open
'bounce' => 0.05,
'feedback' => 0.05
];
$subscribers = $this->subscribersToSend()->get();
if (count($subscribers) == 0) {
throw new Exception('Campaign has ZERO subscribers');
}
echo "Subscriber count: ".count($subscribers)."\n";
$params = array_merge($default, $params);
// Tracking log
foreach ($subscribers as $subscriber) {
// Pick up an available sending server
// Throw exception in case no server available
$server = $this->pickSendingServer($this);
$customerUid = $this->customer->uid;
$msgId = StringHelper::generateMessageId(StringHelper::getDomainFromEmail($this->from_email), $customerUid);
$sent = new \App\SendingServers\Drivers\SendResult(
runtimeMessageId: $msgId,
status: \App\SendingServers\Drivers\DeliveryStatus::SENT,
);
$this->trackMessage($sent, $subscriber, $server, $msgId);
// additional log
}
echo "All tracking log count: ". $this->trackingLogs()->count()."\n";
// Failed tracking log
$failureCount = round($params['delivery_failed'] * $this->trackingLogs()->count());
if ($failureCount == 0) {
throw new Exception("failureCount must be larger than 0.");
}
$allLogs = $this->trackingLogs()->get()->map(function ($e) {
return $e->id;
})->toArray();
$failedLogs = array_values(array_intersect_key($allLogs, array_flip($this->arrayRandFixed($allLogs, $failureCount))));
echo "Failed log count: ".$this->trackingLogs()->whereIn('id', $failedLogs)->count()."\n";
$this->trackingLogs()->whereIn('id', $failedLogs)->update([
'status' => 'failed',
'error' => 'RFC8343: mailbox does not exist',
]);
// Open log
$sentCount = $this->trackingLogs()->where('status', 'sent')->count();
$openCount = round($params['open'] * $sentCount);
$sentLogs = $this->trackingLogs()->where('status', 'sent')
->select('tracking_logs.id')
->get()->map(function ($e) {
return $e->id;
})->toArray();
$openLogs = array_values(array_intersect_key($sentLogs, array_flip($this->arrayRandFixed($sentLogs, $openCount))));
echo "Open log count: ".$openCount."\n";
foreach ($this->trackingLogs()->whereIn('id', $openLogs)->get() as $r) {
$log = new \App\Model\OpenLog();
$log->message_id = $r->message_id;
$log->tracking_log_id = $r->id;
$location = \App\Model\IpLocation::add(StringHelper::getRandomUSIpAddresses());
$log->ip_address = $location->ip_address;
$log->user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36';
$log->setConnection($this->customer->getDbConnection())->save();
}
// Open log
$clickCount = round($params['click'] * $openCount);
$openLogs = $this->trackingLogs()
->join('open_logs', 'tracking_logs.message_id', '=', 'open_logs.message_id')
->select('tracking_logs.id')
->get()->map(function ($e) {
return $e->id;
})->toArray();
$clickLogs = array_values(array_intersect_key($openLogs, array_flip($this->arrayRandFixed($openLogs, $clickCount))));
echo "Click log count: ".$this->trackingLogs()->whereIn('id', $clickLogs)->count()."\n";
foreach ($this->trackingLogs()->whereIn('id', $clickLogs)->get() as $r) {
$log = new \App\Model\ClickLog();
$log->message_id = $r->message_id;
$log->tracking_log_id = $r->id;
$location = \App\Model\IpLocation::add(StringHelper::getRandomUSIpAddresses());
$log->ip_address = $location->ip_address;
$log->user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36';
$log->url = 'https://app.clickfunnels.com/signupflow';
$result = $log->setConnection($this->customer->getDbConnection())->save();
}
// Unsubscribe log
$unsubscribeCount = round($params['unsubscribe'] * $openCount);
$unsubscribeLogs = array_values(array_intersect_key($openLogs, array_flip($this->arrayRandFixed($openLogs, $unsubscribeCount))));
echo "Unsubscribe log count: ".$unsubscribeCount."\n";
foreach ($this->trackingLogs()->whereIn('id', $unsubscribeLogs)->get() as $r) {
// Unsubscribe log
$log = new \App\Model\UnsubscribeLog();
$log->customer_id = $this->customer_id;
$log->message_id = $r->message_id;
$log->subscriber_id = $r->subscriber_id;
$location = \App\Model\IpLocation::add(StringHelper::getRandomUSIpAddresses());
$log->ip_address = $location->ip_address;
$log->user_agent = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36';
$log->setConnection($this->customer->getDbConnection())->save();
}
// Bounce log
$bounceCount = round($params['bounce'] * $sentCount);
$bounceLogs = array_values(array_intersect_key($sentLogs, array_flip($this->arrayRandFixed($sentLogs, $bounceCount))));
echo "Bounce log count: ".$bounceCount."\n";
foreach ($this->trackingLogs()->whereIn('id', $bounceLogs)->get() as $r) {
// Unsubscribe log
$bounceLog = new BounceLog();
$bounceLog->runtime_message_id = $r->message_id;
$bounceLog->message_id = $r->message_id;
$bounceLog->tracking_log_id = $r->id;
// SendGrid only notifies in case of HARD bounce
$bounceLog->bounce_type = BounceLog::HARD;
$bounceLog->raw = 'RFC308433: mailbox is no longer valid or blocked';
$bounceLog->setConnection($this->customer->getDbConnection())->save();
// add subscriber's email to blacklist
$subscriber = $bounceLog->findSubscriberByRuntimeMessageId();
$subscriber->sendToBlacklist($bounceLog->raw);
}
// Bounce log
$feedbackCount = round($params['feedback'] * $sentCount);
$feedbackLogs = array_values(array_intersect_key($sentLogs, array_flip($this->arrayRandFixed($sentLogs, $feedbackCount))));
echo "Feedback log count: ".$feedbackCount."\n";
foreach ($this->trackingLogs()->whereIn('id', $feedbackLogs)->get() as $r) {
// Unsubscribe log
$feedback = new FeedbackLog();
$feedback->runtime_message_id = $r->message_id;
$feedback->message_id = $r->message_id;
$feedback->tracking_log_id = $r->id;
$feedback->feedback_type = 'spam';
$feedback->raw_feedback_content = 'RFC308433: mailbox is no longer valid or blocked';
$feedback->setConnection($this->customer->getDbConnection())->save();
// add subscriber's email to blacklist
$subscriber = $feedback->findSubscriberByRuntimeMessageId();
$subscriber->markAsSpamReported($feedback->raw);
}
// Adjust date/time
$sample = [
'subscriber_first_created' => '6 months ago',
'subscriber_last_created' => '2 day ago',
'first_open' => '24 hours ago',
'last_open' => '2 minutes ago',
];
foreach ($sample as $key => $value) {
$sample[$key] = new Carbon($value);
}
DB::update(sprintf(
'UPDATE %s SET created_at = TIMESTAMPADD(SECOND, FLOOR(RAND() * TIMESTAMPDIFF(SECOND, \'%s\', \'%s\')), \'%s\')',
table('subscribers'),
$sample['subscriber_first_created'],
$sample['subscriber_last_created'],
$sample['subscriber_first_created']
));
DB::update(sprintf(
'UPDATE %s SET created_at = TIMESTAMPADD(SECOND, FLOOR(RAND() * TIMESTAMPDIFF(SECOND, \'%s\', \'%s\')), \'%s\')',
table('open_logs'),
$sample['first_open'],
$sample['last_open'],
$sample['first_open']
));
DB::update(sprintf(
'UPDATE %s SET created_at = TIMESTAMPADD(SECOND, FLOOR(RAND() * TIMESTAMPDIFF(SECOND, \'%s\', \'%s\')), \'%s\')',
table('click_logs'),
$sample['first_open'],
$sample['last_open'],
$sample['first_open']
));
DB::update(sprintf(
'UPDATE %s SET created_at = TIMESTAMPADD(SECOND, FLOOR(RAND() * TIMESTAMPDIFF(SECOND, \'%s\', \'%s\')), \'%s\')',
table('bounce_logs'),
$sample['first_open'],
$sample['last_open'],
$sample['first_open']
));
DB::update(sprintf(
'UPDATE %s SET created_at = TIMESTAMPADD(SECOND, FLOOR(RAND() * TIMESTAMPDIFF(SECOND, \'%s\', \'%s\')), \'%s\')',
table('feedback_logs'),
$sample['first_open'],
$sample['last_open'],
$sample['first_open']
));
DB::update(sprintf(
'UPDATE %s SET created_at = TIMESTAMPADD(SECOND, FLOOR(RAND() * TIMESTAMPDIFF(SECOND, \'%s\', \'%s\')), \'%s\')',
table('unsubscribe_logs'),
$sample['first_open'],
$sample['last_open'],
$sample['first_open']
));
// Finalize
AppCache::for($this)->refresh();
$this->setDone();
$this->save();
}
public function cleanSampleData()
{
$this->defaultMailList->subscribers()->update(['status' => 'subscribed']);
Blacklist::query()->delete();
$this->cleanupLogs();
}
public function openUniqDays($start = null, $end = null)
{
$query = $this->openLogs()->select('open_logs.created_at');
$currentTimezone = $this->customer->getTimezone();
if (isset($start)) {
$query = $query->where('open_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('open_logs.created_at', '<=', $end);
}
return $query->orderBy('open_logs.created_at', 'asc')->get()->groupBy(function ($date) use ($currentTimezone) {
return $date->created_at->timezone($currentTimezone)->format('Y-m-d'); // grouping by hours
});
}
public function clickDays($start = null, $end = null)
{
$currentTimezone = $this->customer->getTimezone();
$query = $this->clickLogs()->select('click_logs.created_at', 'tracking_logs.subscriber_id');
if (isset($start)) {
$query = $query->where('click_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('click_logs.created_at', '<=', $end);
}
return $query->orderBy('click_logs.created_at', 'asc')->get()->groupBy(function ($date) use ($currentTimezone) {
return $date->created_at->timezone($currentTimezone)->format('Y-m-d'); // grouping by hours
});
}
public function openUniqMonths($start = null, $end = null)
{
$query = $this->openLogs()->select('open_logs.created_at');
$currentTimezone = $this->customer->getTimezone();
if (isset($start)) {
$query = $query->where('open_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('open_logs.created_at', '<=', $end);
}
return $query->orderBy('open_logs.created_at', 'asc')->get()->groupBy(function ($date) use ($currentTimezone) {
return $date->created_at->timezone($currentTimezone)->format('Y-m'); // grouping by months
});
}
public function clickMonths($start = null, $end = null)
{
$currentTimezone = $this->customer->getTimezone();
$query = $this->clickLogs()->select('click_logs.created_at', 'tracking_logs.subscriber_id');
if (isset($start)) {
$query = $query->where('click_logs.created_at', '>=', $start);
}
if (isset($end)) {
$query = $query->where('click_logs.created_at', '<=', $end);
}
return $query->orderBy('click_logs.created_at', 'asc')->get()->groupBy(function ($date) use ($currentTimezone) {
return $date->created_at->timezone($currentTimezone)->format('Y-m'); // grouping by months
});
}
public function setPreheader($preheader)
{
$this->preheader = $preheader;
$this->save();
}
public function removePreheader()
{
$this->preheader = null;
$this->save();
}
public function newCampaignHeader()
{
$campaignHeader = campaignHeader::newDefault();
$campaignHeader->campaign_id = $this->id;
$campaignHeader->customer_id = $this->customer_id;
return $campaignHeader;
}
public function useSendingServerDefaultFromEmailAddress()
{
return $this->use_default_sending_server_from_email == true;
}
public function fillDeliveryStatuses(array $array)
{
$array = array_map(function ($status) {
$status = trim($status);
//
if (empty($status)) {
throw new \Exception("Status can not be empty");
}
//
if (!in_array($status, [
\App\Model\Subscriber::VERIFICATION_STATUS_DELIVERABLE,
\App\Model\Subscriber::VERIFICATION_STATUS_UNDELIVERABLE,
\App\Model\Subscriber::VERIFICATION_STATUS_UNKNOWN,
\App\Model\Subscriber::VERIFICATION_STATUS_RISKY,
\App\Model\Subscriber::VERIFICATION_STATUS_UNVERIFIED,
])) {
throw new \Exception("Status $status is invalid!");
}
return $status;
}, $array);
$this->delivery_statuses = json_encode($array);
}
public function setDeliveryStatuses(array $array)
{
$this->fillDeliveryStatuses($array);
$this->save();
}
public function getDefaultDeliveryStatuses()
{
return [
\App\Model\Subscriber::VERIFICATION_STATUS_DELIVERABLE,
\App\Model\Subscriber::VERIFICATION_STATUS_UNVERIFIED,
];
}
public function getDeliveryStatuses()
{
if ($this->delivery_statuses == null) {
// Default delivery statuses
return $this->getDefaultDeliveryStatuses();
}
return json_decode($this->delivery_statuses, true);
}
public function isPlain()
{
return $this->type == Email::TYPE_PLAIN_TEXT;
}
public function archive()
{
// This action is kept out of the transaction to prevent making a too large transaction
// i.e. delete thousands of rows and inserting other thousands of rows)
$this->clearArchives();
$this->doArchive();
$this->cleanupLogs();
$this->setArchiveStatusYes();
}
public function clearArchives()
{
$this->campaignArchives()->delete();
}
private function doArchive()
{
\DB::table('campaign_archives')->insertUsing(
[
'customer_id',
'campaign_id',
'email_id',
'subscriber_id',
'sending_server_id',
'runtime_message_id',
'message_id',
'status',
'error',
'created_at',
'updated_at',
'open_at',
'open_ip_address',
'open_user_agent',
'click_at',
'click_url',
'click_ip_address',
'click_user_agent',
'bounce_at',
'bounce_status_code',
'bounce_type',
'bounce_raw',
'feedback_at',
'feedback_type',
'feedback_raw',
'unsubscribe_at',
'unsubscribe_ip_address',
'unsubscribe_user_agent'
],
$this->subscribers()
->leftJoinSub(
$this->trackingLogs()->getQuery(),
'tracking_logs',
function ($join) {
$join->on('tracking_logs.subscriber_id', 'subscribers.id');
}
)
->leftJoin('bounce_logs', 'tracking_logs.id', 'bounce_logs.tracking_log_id')
->leftJoin('feedback_logs', 'tracking_logs.id', 'feedback_logs.tracking_log_id')
->leftJoin('unsubscribe_logs', 'tracking_logs.message_id', 'unsubscribe_logs.message_id')
->leftJoin('open_logs', 'tracking_logs.id', 'open_logs.tracking_log_id')
->leftJoin('click_logs', 'tracking_logs.id', 'click_logs.tracking_log_id')
->select([
DB::raw($this->customer_id),
'tracking_logs.campaign_id',
'subscribers.id as subscriber_id',
'tracking_logs.sending_server_id',
'tracking_logs.runtime_message_id',
'tracking_logs.message_id',
DB::raw("
CASE
WHEN bounce_logs.id IS NOT NULL THEN '" . self::DELIVERY_STATUS_BOUNCED . "'
WHEN feedback_logs.id IS NOT NULL THEN '" . self::DELIVERY_STATUS_FEEDBACK . "'
WHEN unsubscribe_logs.id IS NOT NULL THEN '" . self::DELIVERY_STATUS_UNSUBSCRIBED . "'
ELSE
CASE tracking_logs.status
WHEN '" . self::DELIVERY_STATUS_SENT . "' THEN '" . self::DELIVERY_STATUS_SENT . "'
WHEN '" . self::DELIVERY_STATUS_FAILED . "' THEN '" . self::DELIVERY_STATUS_FAILED . "'
ELSE
CASE subscribers.status
WHEN '" . Subscriber::STATUS_SUBSCRIBED . "' THEN
CASE COALESCE(subscribers.verification_status, '-1')
WHEN '-1' THEN '" . self::DELIVERY_STATUS_NEW . "'
WHEN '" . Subscriber::VERIFICATION_STATUS_DELIVERABLE . "' THEN '" . self::DELIVERY_STATUS_NEW . "'
ELSE '" . self::DELIVERY_STATUS_SKIPPED . "'
END
ELSE '" . self::DELIVERY_STATUS_SKIPPED . "'
END
END
END AS status
"),
DB::raw('tracking_logs.error as error'),
DB::raw('tracking_logs.created_at as created_at'),
DB::raw('tracking_logs.updated_at as updated_at'),
'open_logs.created_at as open_at',
'open_logs.ip_address as open_ip_address',
'open_logs.user_agent as open_user_agent',
'click_logs.created_at as click_at',
'click_logs.url as click_url',
'click_logs.ip_address as click_ip_address',
'click_logs.user_agent as click_user_agent',
'bounce_logs.created_at as bounce_at',
'bounce_logs.status_code as bounce_status_code',
'bounce_logs.bounce_type as bounce_type',
'bounce_logs.raw as bounce_raw',
'feedback_logs.created_at as feedback_at',
'feedback_logs.feedback_type as feedback_type',
'feedback_logs.raw_feedback_content as feedback_raw',
'unsubscribe_logs.created_at as unsubscribe_at',
'unsubscribe_logs.ip_address as unsubscribe_ip_address',
'unsubscribe_logs.user_agent as unsubscribe_user_agent'
])
);
}
public function archivedSubscribers()
{
return $this->belongsToMany(Subscriber::class, 'campaign_archives');
}
// Interface as App\Library\Contracts\TagResolverInterface
public function getResolvedTags(): array
{
return [
'campaign' => [
'name' => $this->name,
'uid' => $this->uid,
'from_email' => $this->from_email,
'from_name' => $this->from_name,
'subject' => $this->subject,
'reply_to' => $this->reply_to,
],
];
}
// Campaign lifecycle status (is_error / is_paused are boolean flag overlays)
public const STATUS_NEW = 'new';
public const STATUS_QUEUED = 'queued';
public const STATUS_SENDING = 'sending';
public const STATUS_DONE = 'done';
public const STATUS_SCHEDULED = 'scheduled';
public const STATUS_AWAITING_WINNER = 'awaiting_winner';
public const STATUS_SENDING_WINNER = 'sending_winner';
public const ARCHIVE_STATUS_IN_PROGRESS = 'in-progress';
public const ARCHIVE_STATUS_YES = 'yes';
public const ARCHIVE_STATUS_ERROR = 'error';
public const JOB_TYPE_RUN_CAMPAIGN = 'run-campaign';
public const JOB_TYPE_DISPATCH_AND_SEND_MESSAGES = 'dispatch-and-send-messages';
/**
* @return \Illuminate\Database\Eloquent\Relations\BelongsTo<Email, $this>
*/
public function email(): \Illuminate\Database\Eloquent\Relations\BelongsTo
{
return $this->belongsTo(Email::class);
}
/**
* @return \Illuminate\Database\Eloquent\Relations\HasOneThrough<\App\Model\Template, Email, $this>
*/
public function template(): \Illuminate\Database\Eloquent\Relations\HasOneThrough
{
return $this->hasOneThrough(
\App\Model\Template::class,
Email::class,
'id', // emails.id
'id', // templates.id
'email_id', // campaigns.email_id
'template_id' // emails.template_id
);
}
/**
* @return \Illuminate\Database\Eloquent\Relations\HasOne<AbTest, $this>
*/
public function abTest(): \Illuminate\Database\Eloquent\Relations\HasOne
{
return $this->hasOne(AbTest::class);
}
public function hasAbTest(): bool
{
return $this->abTest()->exists();
}
/**
* Override save() to:
* 1. Auto-create an Email if EMAIL_ATTRIBUTES were set before email_id existed (e.g. newDefault()).
* 2. Persist any dirty Email attributes set via delegation.
*/
public function save(array $options = []): bool
{
$pending = array_intersect_key($this->attributes, array_flip(static::EMAIL_ATTRIBUTES));
if (!empty($pending)) {
if (is_null($this->attributes['email_id'] ?? null)) {
// First save — no Email yet: create one and move pending attrs into it
$email = Email::newDefault();
$email->customer_id = $this->attributes['customer_id'] ?? null;
foreach ($pending as $key => $value) {
$email->setAttribute($key, $value);
}
$email->save();
$this->attributes['email_id'] = $email->id;
$this->setRelation('email', $email);
} else {
// email_id is already set — delegate pending attrs to the existing Email
$email = $this->getRelationValue('email') ?? $this->email;
foreach ($pending as $key => $value) {
$email->setAttribute($key, $value);
}
}
// Remove ghost attrs from Campaign's own attribute bag in both cases
foreach (array_keys($pending) as $key) {
unset($this->attributes[$key]);
}
}
$result = parent::save($options);
if ($this->relationLoaded('email') && $this->email && $this->email->isDirty()) {
$this->email->save();
}
return $result;
}
public function customer()
{
return $this->belongsTo('App\Model\Customer');
}
public function scopeScheduled($query)
{
return $query->where('status', static::STATUS_SCHEDULED);
}
public function scopeQueued($query)
{
return $query->where('status', static::STATUS_QUEUED);
}
public function scopeSending($query)
{
return $query->where('status', static::STATUS_SENDING);
}
public function scopeSendingOrQueued($query)
{
return $query->whereIn('status', [static::STATUS_SENDING, static::STATUS_QUEUED]);
}
public function setDone()
{
$this->is_error = false;
$this->is_paused = false;
$this->status = self::STATUS_DONE;
$this->last_error = null;
$this->save();
}
public function setSending()
{
$this->is_error = false;
$this->is_paused = false;
$this->status = self::STATUS_SENDING;
$this->running_pid = getmypid();
$this->delivery_at = Carbon::now();
$this->save();
}
public function isSending()
{
return $this->status == self::STATUS_SENDING;
}
public function isDone()
{
return $this->status == self::STATUS_DONE;
}
public function isAwaitingWinner()
{
return $this->status == self::STATUS_AWAITING_WINNER;
}
public function isSendingWinner()
{
return $this->status == self::STATUS_SENDING_WINNER;
}
public function isQueued()
{
return $this->status == self::STATUS_QUEUED;
}
public function setQueued()
{
$this->is_error = false;
$this->is_paused = false;
$this->status = self::STATUS_QUEUED;
$this->save();
return $this;
}
public function setError($message, $backtrace = null)
{
$this->is_error = true;
$this->setLastError($message, $backtrace);
$this->save();
return $this;
}
public function deleteAndCleanup()
{
if ($this->template) {
$this->template->deleteAndCleanup();
}
$this->cancelAndDeleteJobs();
$this->cleanupLogs();
$this->delete();
}
public function isError()
{
return (bool) $this->is_error;
}
/**
* Plan-level launch policy: returns a translated error string when the
* customer's plan blocks this campaign from launching, or null when OK.
*
* Sources of truth:
* - REQUIRES_TRACKING_DOMAIN entitlement → campaign must have tracking_domain_id
* - HAS_UNSUBSCRIBE_URL_REQUIRED is enforced at template-content save time,
* not here, so it's NOT duplicated.
*
* Call at every entry that flips a campaign into SENDING/QUEUED state
* (wizard launch, A/B launch, API run, RunCampaign re-check).
*/
public function validateLaunchPolicy(?\App\Model\Customer $customer = null): ?string
{
$customer = $customer ?? $this->customer;
if ($customer === null) {
return null;
}
if (\App\Services\Plans\Gates\EntitlementGate::allows($customer, \App\Services\Plans\Entitlements\EntitlementKey::REQUIRES_TRACKING_DOMAIN)
&& $this->tracking_domain_id === null) {
return trans('messages.campaign.tracking_domain.required');
}
return null;
}
public function execute($force = false, string $queue = ACM_QUEUE_TYPE_BATCH)
{
if (!empty($this->customer->custom_queue_name)) {
$queue = $this->customer->custom_queue_name;
}
$shouldDispatch = false;
$alreadyRunning = false;
try {
// Atomic transition:
// read latest status -> decide -> set QUEUED
// Keep dispatch outside lock to avoid deadlocks with sync queue driver.
$this->withLock(function () use ($force, &$shouldDispatch, &$alreadyRunning) {
$this->refresh();
$now = Carbon::now();
if (!is_null($this->run_at) && $this->run_at->gte($now)) {
$scheduledAt = $this->run_at->timezone($this->customer->timezone);
$this->logger()->warning(sprintf('Campaign is scheduled at %s (%s)', $this->customer->formatDateTime($scheduledAt, 'datetime_short'), $scheduledAt->diffForHumans()));
return;
}
if ($this->isSending() || $this->isQueued()) {
if (!$force) {
$alreadyRunning = true;
return;
}
$this->logger()->warning('Force running campaign');
}
DB::transaction(function () {
$this->cancelAndDeleteJobs($jobType = static::JOB_TYPE_RUN_CAMPAIGN);
$this->cancelAndDeleteJobs($jobType = static::JOB_TYPE_DISPATCH_AND_SEND_MESSAGES);
// For A/B tests: populate/top-up subscriber assignments before each send.
// Safe to call on resume — ExactSplitStrategy only assigns unassigned
// subscribers and preserves the existing split ratio.
if ($this->hasAbTest()) {
\App\Jobs\PopulateAbTestAssignments::dispatchSync($this->abTest->id);
}
// Set QUEUED before dispatching so the status is visible even when
// the queue driver runs the job synchronously (sync driver in tests).
$this->setQueued();
});
$shouldDispatch = true;
}, $waitFor = 15, $lockFor = 300);
} catch (\Throwable $e) {
// The DB::transaction() rolled back. Record the error outside the
// transaction so it persists regardless of what was rolled back.
$this->logger()->error('Campaign execute failed: ' . $e->getMessage());
$this->refresh();
if (!$this->isDone() && !$this->isError()) {
$this->setError(
'Failed to start campaign: ' . $e->getMessage(),
substr($e->getTraceAsString(), 0, 8000)
);
}
return;
}
if ($alreadyRunning) {
throw new Exception('Cannot execute: campaign is already in "sending" or "queued" status');
}
if (!$shouldDispatch) {
return;
}
try {
$job = (new RunCampaign($this, $queue));
$this->dispatchWithMonitor($job, $jobType = static::JOB_TYPE_RUN_CAMPAIGN, $queue);
} catch (\Throwable $e) {
$this->logger()->error('Campaign execute dispatch failed: ' . $e->getMessage());
$this->refresh();
if (!$this->isDone() && !$this->isError()) {
$this->setError(
'Failed to dispatch campaign: ' . $e->getMessage(),
substr($e->getTraceAsString(), 0, 8000)
);
}
}
}
public function setScheduled()
{
$this->is_error = false;
$this->is_paused = false;
$this->status = self::STATUS_SCHEDULED;
$this->save();
return $this;
}
public function resume(string $queue)
{
$this->logger()->warning('Resume campaign in queue: '.$queue);
if ($this->is_paused || $this->is_error) {
$this->is_paused = false;
$this->is_error = false;
$this->last_error = null;
$this->save();
}
if ($this->isSendingWinner()) {
// Winner rollout interrupted → continue sending winner
$this->run($check = false, $queue);
} elseif ($this->isAwaitingWinner()) {
// Awaiting winner — nothing to resend, just clear flags.
// UI will allow evaluate/choose winner.
} elseif ($this->hasAbTest() && $this->isSending()) {
// A/B test phase interrupted → continue sending test variants.
// Use run() directly instead of execute() to:
// - preserve 'sending' status (no regression to 'queued')
// - avoid cancelling entire batch + full restart cycle
// - buildAbTestLoaders() picks up only unsent subscribers via tracking_logs dedup
$this->run($check = false, $queue);
} else {
// Normal campaign or non-sending state → full restart via execute()
$this->execute($force = true, $queue);
}
}
public function run($check = true, string $queue = ACM_QUEUE_TYPE_BATCH)
{
$alreadyRunning = false;
$hasActiveDispatchBatch = false;
// Atomic transition:
// refresh -> status/monitor checks -> set SENDING
$this->withLock(function () use ($check, &$alreadyRunning, &$hasActiveDispatchBatch) {
$this->refresh();
$hasActiveDispatchBatch = $this->jobMonitors()
->byJobType(static::JOB_TYPE_DISPATCH_AND_SEND_MESSAGES)
->active()
->exists();
if ($hasActiveDispatchBatch) {
if ($check) {
$alreadyRunning = true;
}
return;
}
if ($check && $this->isSending()) {
$alreadyRunning = true;
return;
}
if (!$check) {
$this->logger()->info('Run without status check');
}
// Preserve SENDING_WINNER status — the batch finally callback uses it
// to know when the winner rollout phase has completed.
if ($this->isSendingWinner()) {
$this->is_error = false;
$this->is_paused = false;
$this->running_pid = getmypid();
$this->save();
} else {
$this->setSending();
}
}, $waitFor = 15, $lockFor = 120);
if ($alreadyRunning) {
throw new Exception('Campaign is already in progress');
}
if ($hasActiveDispatchBatch) {
$this->logger()->warning('Run skipped: active dispatch batch already exists');
return;
}
$jobs = $this->jobMonitors()->byJobType(static::JOB_TYPE_DISPATCH_AND_SEND_MESSAGES)->get();
foreach ($jobs as $job) {
$job->cancelWithoutDeleteBatch();
}
$this->setDelayFlag(null);
$this->debug(function ($info) {
$info['delay_note'] = null;
return $info;
});
$campaignLoaders = $this->hasAbTest()
? $this->buildAbTestLoaders()
: $this->buildNormalLoaders();
$campaignId = $this->id;
$className = get_called_class();
$customerId = $this->customer->id;
$monitor = $this->dispatchWithBatchMonitor(
$jobTypeName = static::JOB_TYPE_DISPATCH_AND_SEND_MESSAGES,
$queue,
$campaignLoaders,
function ($batch) use ($campaignId, $className, $queue, $customerId) {
\App\Jobs\ContinueCampaign::dispatch($campaignId, $className, $queue, $customerId)
->onQueue($queue);
},
function (Batch $batch, Throwable $e) use ($campaignId, $className, $customerId) {
$customer = \App\Model\Customer::find($customerId);
$campaign = $className::on($customer->getDbConnection())->find($campaignId);
$errorMsg = "Campaign stopped. ".$e->getMessage()."\n".$e->getTraceAsString();
$campaign->logger()->info($errorMsg);
$campaign->setError(
'Campaign stopped. '.$e->getMessage(),
substr($e->getTraceAsString(), 0, 8000)
);
},
function () use ($campaignId, $className, $customerId) {
$customer = \App\Model\Customer::find($customerId);
$campaign = $className::on($customer->getDbConnection())->find($campaignId);
$updateCacheJobId = uniqid();
$campaign->logger()->info('Finally callback of batch! Updating cache '.$updateCacheJobId);
AppCache::for($campaign)->refresh();
$campaign->logger()->info('DONE updating cache '.$updateCacheJobId);
// Status transitions (setDone, AWAITING_WINNER) are handled
// by ContinueCampaign — single source of truth.
// This callback only handles cache update and cleanup.
if (!$campaign->hasAbTest() && !Setting::isYes('campaign.duplicate')) {
$campaign->cleanupDuplicateTable();
}
}
);
$this->logger()->info('Batch dispatched, JobMonitor #'.$monitor->id);
}
/**
* Build LoadCampaign jobs for a normal (non-A/B) campaign.
* Extracted from the original run() body.
*/
private function buildNormalLoaders(): array
{
$perPage = 100;
$maxPageToLoad = 2;
$campaignLoaders = [];
$pageCount = 0;
if (Setting::isYes('campaign.duplicate')) {
$subscribersQuery = $this->subscribersToSend();
} else {
$this->calculateDuplicateEmails();
$subscribersQuery = $this->subscribersToSend()->leftJoin($this->getDuplicateTable(), function ($join) {
$join->on("{$this->getDuplicateTable()}.email", '=', 'subscribers.email');
})->where(function ($query) {
$query->whereNull("{$this->getDuplicateTable()}.email")->orWhere("{$this->getDuplicateTable()}.selected_id", '=', DB::raw('subscribers.id'));
});
$campaignLoaders[] = new HandleDuplicateEmails($this);
}
paginate_query($subscribersQuery, $perPage, $orderBy = null, function ($pageNumber, $subscribers) use (&$campaignLoaders, &$pageCount) {
$pageCount += 1;
$listOfIds = $subscribers->pluck('subscribers.id')->toArray();
$campaignLoaders[] = new LoadCampaign($this, $pageNumber, $listOfIds);
}, $maxPageToLoad);
if ($pageCount == 0) {
$campaignLoaders[] = new LoadCampaign($this, 0, []);
}
return $campaignLoaders;
}
/**
* Build LoadCampaign jobs for an A/B test campaign.
* Each variant gets its own set of jobs; subscribers are filtered via ab_test_assignments JOIN.
*/
private function buildAbTestLoaders(): array
{
$abTest = $this->abTest()->with('variants.email')->firstOrFail();
$perPage = 100;
$maxPageToLoad = 2;
$campaignLoaders = [];
$pageCount = 0;
foreach ($abTest->variants as $variant) {
// Filter: only subscribers assigned to this variant
$subscribersQuery = $this->subscribersToSend()
->join('ab_test_assignments as ata', function ($join) use ($abTest, $variant) {
$join->on('ata.subscriber_id', '=', 'subscribers.id')
->where('ata.ab_test_id', '=', $abTest->id)
->where('ata.variant_id', '=', $variant->id);
});
paginate_query($subscribersQuery, $perPage, $orderBy = null, function ($pageNumber, $subscribers) use (&$campaignLoaders, &$pageCount, $variant) {
$pageCount += 1;
$listOfIds = $subscribers->pluck('subscribers.id')->toArray();
$campaignLoaders[] = new LoadCampaign($this, $pageNumber, $listOfIds, $variant);
}, $maxPageToLoad);
}
if ($pageCount === 0) {
// No subscribers yet — still need one loader so the batch can close and ContinueCampaign fires
$campaignLoaders[] = new LoadCampaign($this, 0, [], $abTest->variants->first());
}
return $campaignLoaders;
}
public function logger()
{
if (!is_null($this->logger)) {
return $this->logger;
}
$formatter = new LineFormatter("[%datetime%] %channel%.%level_name%: %message%\n");
$logfile = $this->getLogFile();
$stream = new RotatingFileHandler($logfile, 7, config('custom.log_level'));
$stream->setFormatter($formatter);
$pid = getmypid();
$logger = new MonologLogger($pid);
$logger->pushHandler($stream);
$this->logger = $logger;
return $this->logger;
}
public function getLogFile()
{
$path = storage_path(join_paths('logs', php_sapi_name(), '/campaign-'.$this->uid.'.log'));
return $path;
}
public function extractErrorMessage()
{
return $this->last_error;
}
public function scheduleDiffForHumans()
{
if ($this->run_at) {
return $this->run_at->timezone($this->customer->timezone)->diffForHumans();
} else {
return null;
}
}
public function pause()
{
$this->cancelAndDeleteJobs();
$this->setPaused();
$this->logger()->warning('Campaign paused by user');
event(new CampaignUpdated($this));
}
public function setFailed($error)
{
$this->cancelAndDeleteJobs();
$this->setError($error);
$this->logger()->warning('Campaign was set as failed. '.$error);
event(new CampaignUpdated($this));
}
public function setPaused()
{
$this->is_paused = true;
$this->save();
return $this;
}
public function isPaused()
{
return (bool) $this->is_paused;
}
public function scopeByStatus($query, $status)
{
// Handle boolean flag filters
if ($status === 'is_error') {
return $query->where('is_error', true);
}
if ($status === 'is_paused') {
return $query->where('is_paused', true);
}
return $query->where('status', $status);
}
public static function statusSelectOptions()
{
return [
['text' => trans('messages.campaign_status_' . self::STATUS_NEW), 'value' => self::STATUS_NEW],
['text' => trans('messages.campaign_status_' . self::STATUS_QUEUED), 'value' => self::STATUS_QUEUED],
['text' => trans('messages.campaign_status_' . self::STATUS_SENDING), 'value' => self::STATUS_SENDING],
['text' => trans('messages.campaign_status_' . self::STATUS_DONE), 'value' => self::STATUS_DONE],
['text' => trans('messages.campaign_status_' . self::STATUS_SCHEDULED), 'value' => self::STATUS_SCHEDULED],
['text' => trans('messages.campaign_status_' . self::STATUS_AWAITING_WINNER), 'value' => self::STATUS_AWAITING_WINNER],
['text' => trans('messages.campaign_status_' . self::STATUS_SENDING_WINNER), 'value' => self::STATUS_SENDING_WINNER],
['text' => trans('messages.campaign_status_error'), 'value' => 'is_error'],
['text' => trans('messages.campaign_status_paused'), 'value' => 'is_paused'],
];
}
private function getDebugCacheKey()
{
return $key = 'debug-campaign-'.$this->uid;
}
public function cleanupDebug()
{
$key = $this->getDebugCacheKey();
Cache::forget($key);
}
public function debug(Closure $callback = null)
{
$lockKey = "lock-for-debug-campaign-{$this->uid}";
$key = $this->getDebugCacheKey();
if (is_null($callback)) {
return Cache::get($key);
}
$result = null;
with_cache_lock($lockKey, function () use (&$result, $callback, $key) {
$info = Cache::get($key, $default = [
'start_at' => null,
'last_activity_at' => Carbon::now()->toString(),
'finish_at' => null,
'total_time' => null,
'last_message_sent_at' => null,
'messages_sent_per_second' => null,
'send_message_count' => 0,
'send_message_total_time' => 0,
'send_message_prepare_avg_time' => null,
'send_message_lock_avg_time' => null,
'send_message_delivery_avg_time' => null,
'send_message_avg_time' => null,
'send_message_min_time' => null,
'send_message_max_time' => null,
'delay_note' => null,
]);
$result = $callback($info);
Cache::put($key, $result);
}, $timeout = 10);
return $result;
}
/**
* Lightweight liveness ping. Writes the current timestamp to a dedicated
* cache key without acquiring the debug() lock — used by SendPipeline as
* a per-message heartbeat to detect hung jobs. Read via heartbeatAt() for
* stuck-job audits.
*/
public function touchActivity(): void
{
Cache::put("campaign-heartbeat-{$this->uid}", now()->toString());
}
public function heartbeatAt(): ?string
{
return Cache::get("campaign-heartbeat-{$this->uid}");
}
public function getDelayFlagKey()
{
$flagKey = "campaign-delay-flag-{$this->uid}";
return $flagKey;
}
public function checkDelayFlag()
{
return Cache::get($this->getDelayFlagKey()) ?? false;
}
public function setDelayFlag($value)
{
if (is_null($value)) {
Cache::forget($this->getDelayFlagKey());
} else {
Cache::put($this->getDelayFlagKey(), $value);
}
}
public static function scopeByCustomer($query, $customer)
{
$query->where('customer_id', $customer->id);
}
public function withLock(Closure $task, int $waitFor = 15, ?int $lockFor = null)
{
$key = "lock-campaign-{$this->uid}";
return with_cache_lock($key, function () use ($task) {
return $task();
}, $waitFor, $lockFor);
}
public function getDuplicateTable()
{
return '__campaign_duplicate_'.$this->uid;
}
public function cleanupDuplicateTable()
{
$duplicateTable = $this->getDuplicateTable();
$duplicateTableWithPrefix = table($duplicateTable);
DB::statement("DROP TABLE IF EXISTS `{$duplicateTableWithPrefix}`;");
}
public function calculateDuplicateEmails()
{
$duplicateEmails = $this->subscribers()->select('subscribers.email', DB::raw('min(id) AS selected_id'))->groupBy('subscribers.email')->havingRaw('count(*) > 1')->get()->toArray();
$duplicateTable = $this->getDuplicateTable();
$duplicateTableWithPrefix = table($duplicateTable);
DB::statement("DROP TABLE IF EXISTS `{$duplicateTableWithPrefix}`;");
DB::statement("CREATE TABLE `{$duplicateTableWithPrefix}` (`selected_id` int unsigned, `email` VARCHAR(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci) ENGINE=InnoDB;");
DB::table($duplicateTable)->insert($duplicateEmails);
DB::statement("CREATE INDEX `_index_email_{$duplicateTableWithPrefix}` ON `{$duplicateTable}`(`email`);");
DB::statement("UPDATE `{$duplicateTableWithPrefix}` AS dup INNER JOIN `subscribers` s ON dup.email = s.email SET dup.selected_id = s.id WHERE s.mail_list_id = {$this->defaultMailList->id}");
}
public function setArchiveStatusInProgress()
{
$this->archive_status = static::ARCHIVE_STATUS_IN_PROGRESS;
$this->save();
}
public function setArchiveStatusNull()
{
$this->archive_status = null;
$this->save();
}
public function setArchiveStatusYes()
{
$this->archive_status = static::ARCHIVE_STATUS_YES;
$this->save();
}
public function setArchiveStatusError($message, $backtrace = null)
{
$this->setLastError($message, $backtrace);
$this->archive_status = static::ARCHIVE_STATUS_ERROR;
$this->save();
}
public function cleanupLogs()
{
$trackingLogQuery = $this->trackingLogs()->getQuery();
$chunksize = 1000;
$trackingLogQuery->orderBy('id')->chunkById($chunksize, function ($logs) {
$ids = $logs->pluck('id');
$messageIds = $logs->pluck('message_id');
DB::table('open_logs')
->whereIn('tracking_log_id', $ids)
->orWhereIn('message_id', $messageIds)
->delete();
DB::table('click_logs')
->whereIn('tracking_log_id', $ids)
->orWhereIn('message_id', $messageIds)
->delete();
DB::table('bounce_logs')
->whereIn('tracking_log_id', $ids)
->orWhereIn('message_id', $messageIds)
->delete();
DB::table('feedback_logs')
->whereIn('tracking_log_id', $ids)
->orWhereIn('message_id', $messageIds)
->delete();
DB::table('unsubscribe_logs')
->whereIn('message_id', $messageIds)
->delete();
DB::table('tracking_logs')->whereIn('id', $ids)->delete();
});
}
}