PHP code example of borsche / elasticsearch-audit-bundle

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

borsche / elasticsearch-audit-bundle example snippets

Borsche\ElasticsearchAuditBundle\ElasticsearchAuditBundle::class => ['all' => true],

use Borsche\ElasticsearchAuditBundle\Model\Change;
use Borsche\ElasticsearchAuditBundle\Writer\AuditWriter;

final class CallController
{
    public function __construct(private AuditWriter $audit) {}

    public function place(Order $order, Request $request): Response
    {
        // ...
        $this->audit->record(
            objectType: 'order',
            objectId: $order->getId(),
            event: 'order_call',
            changes: ['phone' => $phone, 'duration' => new Change(null, 42)],
        );
    }
}

$this->audit->record('order', 42, 'update', at: $importedAt, actor: 'legacy-import');

use Borsche\ElasticsearchAuditBundle\Attribute\Auditable;
use Borsche\ElasticsearchAuditBundle\Attribute\AuditField;

#[ORM\Entity]
#[Auditable(type: 'article', alwaysRecord: ['status'])]
class Article
{
    #[ORM\Column, AuditField]
    private string $title;

    #[ORM\Column, AuditField]
    private string $status = 'draft';

    #[ORM\Column]
    private int $views = 0;                       // not audited: changes here record nothing

    #[ORM\ManyToOne, AuditField(represent: 'getName')]
    private ?Author $author = null;               // stored as the author's name, not the object

    #[ORM\ManyToMany(targetEntity: Tag::class), AuditField(represent: 'getLabel')]
    private Collection $tags;                     // stored as ['php'] → ['php', 'elasticsearch']
}

use Borsche\ElasticsearchAuditBundle\Contract\AuditableInterface;

class Article implements AuditableInterface
{
    public function getAuditObjectType(): string { return 'article'; }

    public function getAuditedFields(): array
    {
        return [
            'title' => null,                                    // scalar
            'status' => null,
            'author' => fn (Author $a) => $a->getName(),        // to-one, through a representer
            'tags' => fn (Tag $t) => $t->getLabel(),            // to-many, each element represented
        ];
    }

    public function getAlwaysRecordedFields(): array { return ['status']; }
}

$this->frame->begin(atomic: true);
$this->em->getConnection()->beginTransaction();

try {
    // ... several flushes ...
    $this->em->getConnection()->commit();
} catch (\Throwable $e) {
    $this->em->getConnection()->rollBack();
    $this->frame->reset();     // rolled back: drop what never happened
    throw $e;
}

// Committed. The history speaks separately, and outside the try on purpose: a failed
// write here must not reach the rollback above, which would then roll back a
// transaction that is no longer there and bury the real cause under a DBAL error.
$this->frame->end();

#[ORM\OneToMany(mappedBy: 'shipment', targetEntity: ShipmentLine::class)]
#[AuditField(represent: 'getLabel', trackElements: ['quantity'])]   // or trackElements: true
private Collection $lines;

class Shipment implements AuditableInterface, TracksCollectionElementsInterface
{
    public function getTrackedCollections(): array { return ['lines' => ['quantity']]; }
}

use Borsche\ElasticsearchAuditBundle\Coalescing\AuditFrame;

final class MoveStockHandler
{
    public function __construct(private AuditFrame $frame, private StockService $stock) {}

    public function __invoke(MoveStock $command): void
    {
        $this->frame->coalesce(fn () => $this->stock->move($command));
    }
}

use Borsche\ElasticsearchAuditBundle\Model\AuditQuery;
use Borsche\ElasticsearchAuditBundle\Reader\AuditReader;

$page = $this->reader->find(
    AuditQuery::for('order')
        ->withObjectId(42)                          // one object's history...
        ->withEvents('update', 'order_call')        // ...or by event
        ->withActors('7')                           // who
        ->between($since, $until)                   // when (either side may be null)
        ->where('salesType', 3)                     // any attribute an enricher added
        ->whereIn('warehouseId', [1, 2])
        ->whereExists('orderCountry')               // has the attribute (since 0.12)
        ->whereNotExists('legacyRef')               // does not have it — what a backfill looks for
        ->whereBetween('total', 100, 500)           // inclusive range; either bound may be null
        ->page(2, 50)                               // newest first by default; ->oldestFirst()
);

$page->entries;          // list<AuditEntry>: id, objectType, objectId, event, loggedAt, actor, changes, attributes, extra, warnings
$page->total;            // exact
$page->totalPages();
$page->toArray();        // ['items' => [...], 'pagination' => [currentPage, limit, total, totalPages, nextCursor]]

foreach ($page->entries as $entry) {
    if (!$entry->isComplete()) {
        // show it greyed out, leave it out of the CSV, count it in a health check —
        // $entry->warnings[0] is "The logged-at value could not be read, so this entry
        // is dated 1970-01-01. It is not when this happened."
    }
}

$page = $this->reader->find($query->page(1, 100));
// ... later, for the next page:
$next = $this->reader->find($query->after($page->nextCursor()));

$token = (string) $request->query->get('cursor', '');

try {
    $query = AuditQuery::create()->forType('order');
    $query = $token === '' ? $query : $query->afterToken($token);   // an empty token is not a page
    $page = $reader->find($query);                                  // both inside, or the token is not covered
} catch (InvalidQueryException $e) {
    throw new BadRequestHttpException('That page link has expired — start from the first page.', $e);
}

foreach ($this->reader->iterate(AuditQuery::for('order')->since($start)->oldestFirst(), batchSize: 500) as $entry) {
    $sheet->addRow([$entry->loggedAt->format('Y-m-d H:i'), $entry->actor, $entry->event, json_encode($entry->changes)]);
}

use Borsche\ElasticsearchAuditBundle\Contract\QueryExtensionInterface;

final class CountryFilter implements QueryExtensionInterface
{
    public function __construct(private UserRepository $users) {}

    public function extend(AuditQuery $query): AuditQuery
    {
        if (!$query->hasOption('country')) {
            return $query;
        }

        $ids = $this->users->idsInCountry($query->option('country'));

        return $ids === [] ? $query->matchNothing() : $query->narrowActors(...$ids);
    }
}

// in the controller:
$query = AuditQuery::for('order')->withOption('country', $request->query->get('country'));

$response = $this->reader->raw(
    AuditQuery::for('order')->withObjectId(42),
    ['size' => 0, 'aggs' => ['actors' => ['terms' => ['field' => 'source']]]],
);

$buckets = $response['aggregations']['actors']['buckets'] ?? [];   // ?? [] — see below

use Borsche\ElasticsearchAuditBundle\Contract\RecordDecoratorInterface;

final class ActorNames implements RecordDecoratorInterface
{
    public function __construct(private UserRepository $users) {}

    public function decorate(array $entries): array
    {
        $users = $this->users->findIndexedByIds(array_unique(array_filter(array_map(fn ($e) => $e->actor, $entries))));

        return array_map(
            fn (AuditEntry $e) => $e->withExtra(['actor' => $users[$e->actor] ?? null ? ['id' => $e->actor, 'name' => $users[$e->actor]->getName()] : null]),
            $entries,
        );
    }
}

#[Route('/api/history', methods: ['GET'])]
public function history(Request $request, AuditReader $reader): JsonResponse
{
    $query = AuditQuery::for($request->query->getString('objectType', 'order'))
        ->page($request->query->getInt('page', 1), min(100, $request->query->getInt('limit', 20)));

    if ($id = $request->query->get('objectId')) {
        $query = $query->withObjectId($id);
    }

    if ($cursor = $request->query->getString('cursor')) {
        $query = $query->afterToken($cursor); // page numbers no longer apply
    }

    try {
        return $this->json($reader->find($query)->toArray());
    } catch (InvalidQueryException $e) {
        return $this->json(['error' => $e->getMessage()], 400);
    }
}

use Borsche\ElasticsearchAuditBundle\Event\RecordCreatedEvent;
use Borsche\ElasticsearchAuditBundle\Event\RecordFailedEvent;

#[AsEventListener]
final class ShapeTheTrail
{
    public function __invoke(RecordCreatedEvent $event): void
    {
        $record = $event->getRecord();

        if ($record->event === 'heartbeat') {
            $event->veto();          // not written, not an error
        }

        if ($record->objectType === 'order' && !$this->tenants->auditsDetails($record)) {
            $event->setRecord($record->withChanges([]));   // this tenant keeps the fact, not the diff
        }
    }
}

#[AsEventListener]
final class CountAuditFailures
{
    public function __invoke(RecordFailedEvent $event): void
    {
        $this->metrics->increment('audit.write_failed', ['type' => $event->record->objectType]);
    }
}

use Borsche\ElasticsearchAuditBundle\Contract\ActorResolverInterface;

final class ImpersonationActorResolver implements ActorResolverInterface
{
    public function __construct(private ActingUserHolder $holder) {}

    public function resolve(): ?string
    {
        return $this->holder->currentUserId();  // null when unknown → next resolver, then the fallback
    }
}

use Borsche\ElasticsearchAuditBundle\Contract\AuditEnricherInterface;
use Borsche\ElasticsearchAuditBundle\Model\AuditRecord;

final class OrderAttributesEnricher implements AuditEnricherInterface
{
    public function __construct(private OrderRepository $orders) {}

    public function supports(AuditRecord $record): bool
    {
        return $record->objectType === 'order';
    }

    public function enrich(AuditRecord $record): AuditRecord
    {
        $order = $this->orders->find($record->objectId);

        return $record->withAttributes(['salesType' => $order?->getOffer()?->getSalesType()?->getId()]);
    }

    public function mapping(): array
    {
        return ['salesType' => ['type' => 'integer']];
    }
}

use Borsche\ElasticsearchAuditBundle\Contract\ScopedEnricherInterface;

final class OrderAttributesEnricher implements ScopedEnricherInterface
{
    public function objectTypes(): array
    {
        return ['order'];   // [] — or not implementing this — means every type
    }

    public function supports(AuditRecord $record): bool
    {
        return true;        // the object type has already been answered
    }

    // enrich() and mapping() as above
}

final class QuantityChanged implements MergedRecordEnricherInterface
{
    public function supports(AuditRecord $record): bool { return $record->objectType === 'stock'; }

    public function enrich(AuditRecord $record): AuditRecord
    {
        return $record->withAttributes(['quantityChanged' => array_key_exists('quantity', $record->changes)]);
    }

    public function mapping(): array { return ['quantityChanged' => ['type' => 'boolean']]; }
}

final class RouteEnricher implements MomentEnricherInterface
{
    public function __construct(private readonly RequestStack $requests) {}

    public function describe(): array
    {
        $request = $this->requests->getCurrentRequest();

        return $request === null ? [] : ['route' => (string) $request->attributes->get('_route')];
    }

    public function mapping(): array { return ['route' => ['type' => 'keyword']]; }
}

$collector = static::getContainer()->get(AuditCollector::class);

$this->orders->pay($order);

self::assertCount(1, $collector->writtenFor('order', $order->getId(), 'update'), $collector->explain());

$entry = $collector->writtenFor('order', $order->getId())[0]->entry();

self::assertSame('paid', $entry->changes['status']['new']);
self::assertSame('/checkout', $entry->attribute('route'));

use AuditAssertions;

protected function auditCollector(): AuditCollector
{
    return static::getContainer()->get(AuditCollector::class);
}

// Exactly one — a duplicated record is indistinguishable from history, so "at least one" is
// not what a test of an audit trail means.
$record = $this->assertAudited('order', $order->getId(), 'update');

$this->assertAuditVetoed('reminder', $order->getId());
$this->assertNothingAudited();

$this->audit->write($record, immediately: true);

public function __construct(private AuditTransaction $transaction, private EntityManagerInterface $em) {}

public function approve(Order $order): void
{
    $this->transaction->run(function () use ($order): void {
        $order->status = 'approved';
        $this->em->flush();

        $order->totalCents = $this->recompute($order);
        $this->em->flush();
    });
}

// What reached the index in the last hour, whenever it happened. An entry whose loggedAt is
// far behind its writtenAt is one that waited.
$page = $reader->find(AuditQuery::any()->whereBetween('writtenAt', $anHourAgo, null));

foreach ($page->entries as $entry) {
    $waited = strtotime((string) $entry->attribute('writtenAt')) - $entry->loggedAt->getTimestamp();
}

public function resolve(): ?string
{
    $user = $this->tokenStorage->getToken()?->getUser();

    return $user instanceof User ? (string) $user->getId() : null;   // an id, not an email
}

  public function decorate(array $entries): array
  {
      $ids = array_values(array_unique(array_filter(array_map(fn ($e) => $e->objectId, $entries))));
      $orders = [];

      foreach (array_chunk($ids, 500) as $chunk) {
          foreach ($this->orders->findSummaries($chunk) as $row) {   // array rows, not entities
              $orders[$row['id']] = $row;
          }
      }

      return array_map(fn (AuditEntry $e) => $e->withExtra(['order' => $orders[$e->objectId] ?? null]), $entries);
  }
  

bash
curl -X POST "$ES/audit_log/_delete_by_query?conflicts=proceed" -H 'Content-Type: application/json' -d'
{"query": {"range": {"loggedAt": {"lt": "2024-01-01 00:00:00"}}}}'