Import: implement import runs

* Trigger import run and store data
* Add new table showing related information

application/controllers/ImportsourceController.php

@@ -1,8 +1,11 @@
 <?php
 
 use Icinga\Module\Director\Web\Controller\ActionController;
-
 use Icinga\Module\Director\Web\Hook\ImportSourceHook;
+use Icinga\Module\Director\Objects\ImportSource;
+use Icinga\Module\Director\Import\Import;
+use Icinga\Exception\InvalidPropertyException;
+use Icinga\Web\Notification;
 
 class Director_ImportsourceController extends ActionController
 {
@@ -16,6 +19,15 @@ class Director_ImportsourceController extends ActionController
         $this->forward('index', 'importsource', 'director');
     }
 
+    public function runAction()
+    {
+        if ($runId = Import::run($id = ImportSource::load($this->params->get('id'), $this->db()))) {
+            Notification::success('adf' . $runId);
+            $this->redirectNow('director/list/importrun');
+        } else {
+        }
+    }
+
     public function indexAction()
     {
         $edit = false;

application/controllers/ListController.php

@@ -32,12 +32,20 @@ class Director_ListController extends ActionController
             'director/importsource/add'
         );
 
-        $this->setConfigTabs()->activate('importsource');
+        $this->setImportTabs()->activate('importsource');
         $this->view->title = $this->translate('Import source');
         $this->view->table = $this->loadTable('importsource')->setConnection($this->db());
         $this->render('table');
     }
 
+    public function importrunAction()
+    {
+        $this->setImportTabs()->activate('importrun');
+        $this->view->title = $this->translate('Import runs');
+        $this->view->table = $this->loadTable('importrun')->setConnection($this->db());
+        $this->render('table');
+    }
+
     public function datafieldAction()
     {
         $this->view->addLink = $this->view->qlink(

application/tables/ImportrunTable.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace Icinga\Module\Director\Tables;
+
+use Icinga\Module\Director\Web\Table\QuickTable;
+
+class ImportrunTable extends QuickTable
+{
+    public function getColumns()
+    {
+        return array(
+            'id'          => 'r.id',
+            'source_id'   => 's.id',
+            'source_name' => 's.source_name',
+            'start_time'  => 'r.start_time',
+            'rowset'      => 'LOWER(HEX(rs.checksum))',
+            'cnt_rows'    => 'COUNT(rsr.row_checksum)',
+        );
+    }
+
+    protected function getActionUrl($row)
+    {
+        return $this->url('director/now/where', array('id' => $row->id));
+    }
+
+    public function getTitles()
+    {
+        $view = $this->view();
+        return array(
+            'source_name' => $view->translate('Source name'),
+            'start_time'  => $view->translate('Timestamp'),
+            'cnt_rows'    => $view->translate('Imported rows'),
+        );
+    }
+
+    public function fetchData()
+    {
+        $db = $this->connection()->getConnection();
+
+        $query = $db->select()->from(
+            array('s' => 'import_source'),
+            $this->getColumns()
+        )->join(
+            array('r' => 'import_run'),
+            'r.source_id = s.id',
+            array()
+        )->joinLeft(
+            array('rs' => 'imported_rowset'),
+            'rs.checksum = r.imported_rowset_checksum',
+            array()
+        )->joinLeft(
+            array('rsr' => 'imported_rowset_row'),
+            'rs.checksum = rsr.rowset_checksum',
+            array()
+        )->group('r.id')->order('r.start_time DESC');
+
+        return $db->fetchAll($query);
+    }
+}

application/tables/ImportsourceTable.php

@@ -20,6 +20,16 @@ class ImportsourceTable extends QuickTable
         return $this->url('director/importsource/edit', array('id' => $row->id));
     }
 
+    protected function renderAdditionalActions($row)
+    {
+        return $this->view->qlink(
+            'Run',
+            'director/importsource/run',
+            array('id' => $row->id),
+            array('data-base-target' => '_main')
+        );
+    }
+
     public function getTitles()
     {
         $view = $this->view();

configuration.php

@@ -16,6 +16,9 @@ $section->add($this->translate('Global'))->setUrl('director/commands');
 $section->add($this->translate('Hosts'))->setUrl('director/hosts');
 $section->add($this->translate('Services'))->setUrl('director/services');
 $section->add($this->translate('Users'))->setUrl('director/users');
+$section->add($this->translate('Import / Sync'))
+    ->setUrl('director/list/importsource')
+    ->setPriority(901);
 $section->add($this->translate('Config'))
     ->setUrl('director/list/generatedconfig')
     ->setPriority(902);

library/Director/Import/Import.php

@@ -0,0 +1,134 @@
+<?php
+
+namespace Icinga\Module\Director\Import;
+
+use Icinga\Module\Director\Objects\ImportSource;
+use Icinga\Module\Director\Web\Hook\ImportSourceHook;
+
+class Import
+{
+    protected $db;
+
+    protected function __construct()
+    {
+    }
+
+    public static function run(ImportSource $source)
+    {
+        $import = new self();
+        return $import->importFromSource($source);
+    }
+
+    protected function importFromSource(ImportSource $source)
+    {
+        $connection = $source->getConnection();
+        $this->db = $db = $connection->getDbAdapter();
+
+        $keyColumn = $source->key_column;
+        $rows = array();
+        $props = array();
+        $rowsProps = array();
+
+        foreach (ImportSourceHook::loadByName($source->source_name, $connection)->fetchData() as $row) {
+            // TODO: Check for name collision
+            if (! isset($row->$keyColumn)) {
+                die('Depp');
+            }
+            $object_name = $row->$keyColumn;
+            $rowChecksums = array();
+            $keys = array_keys((array) $row);
+            sort($keys);
+
+            foreach ($keys as $key) {
+                $checksum = sha1($key . '=' . json_encode((string) $row->$key), true);
+
+                if (! array_key_exists($checksum, $props)) {
+                    $props[$checksum] = array(
+                        'checksum'       => $checksum,
+                        'property_name'  => $key,
+                        'property_value' => $row->$key,
+                        'format'         => 'string'
+                    );
+                }
+
+                $rowChecksums[] = $checksum;
+            }
+
+            $checksum = sha1($object_name . ';' . implode(';', $rowChecksums), true);
+            if (array_key_exists($checksum, $rows)) {
+                die('WTF, collision?');
+            }
+
+            $rows[$checksum] = array(
+                'checksum'    => $checksum,
+                'object_name' => $object_name
+            );
+
+            $rowsProps[$checksum] = $rowChecksums;
+        }
+
+        $rowSums = array_keys($rows);
+        $rowset = sha1(implode(';', $rowSums), true);
+
+        $db->beginTransaction();
+        if (! $this->rowSetExists($rowset)) {
+
+            $newRows = $this->newChecksums('imported_row', $rowSums);
+            $newProperties = $this->newChecksums('imported_property', array_keys($props));
+
+            if (! empty($newProperties) || ! empty($newRows)) {
+                foreach ($newProperties as $checksum) {
+                    $db->insert('imported_property', $props[$checksum]);
+                }
+
+                $db->insert('imported_rowset', array('checksum' => $rowset));
+
+                foreach ($newRows as $checksum) {
+                    $db->insert('imported_row', $rows[$checksum]);
+                    $db->insert(
+                        'imported_rowset_row',
+                        array(
+                            'rowset_checksum' => $rowset,
+                            'row_checksum'    => $checksum
+                        )
+                    );
+                    foreach ($rowsProps[$checksum] as $propChecksum) {
+                        $db->insert('imported_row_property', array(
+                            'row_checksum'      => $checksum,
+                            'property_checksum' => $propChecksum
+                        ));
+                    }
+                }
+            }
+        }
+        $db->insert(
+            'import_run',
+            array(
+                'source_id' => $source->id,
+                'imported_rowset_checksum' => $rowset,
+                'start_time' => date('Y-m-d H:i:s'),
+                'succeeded' => 'y'
+            )
+        );
+        $id =  $db->lastInsertId();
+        $db->commit();
+
+        return $id;
+    }
+
+    protected function rowSetExists($checksum)
+    {
+        return count($this->newChecksums('imported_rowset', array($checksum))) === 0;
+    }
+
+    protected function newChecksums($table, $checksums)
+    {
+        $db = $this->db;
+
+        $existing = $db->fetchCol(
+            $db->select()->from($table, 'checksum')->where('checksum IN (?)', $checksums)
+        );
+
+        return array_diff($checksums, $existing);
+    }
+}

library/Director/Web/Controller/ActionController.php

@@ -48,9 +48,18 @@ abstract class ActionController extends Controller
         )->add('datafield', array(
             'label' => $this->translate('Data fields'),
             'url'   => 'director/list/datafield')
-        )->add('importsource', array(
+        );
+        return $this->view->tabs;
+    }
+
+    protected function setImportTabs()
+    {
+        $this->view->tabs = Widget::create('tabs')->add('importsource', array(
             'label' => $this->translate('Import source'),
             'url'   => 'director/list/importsource')
+        )->add('importrun', array(
+            'label' => $this->translate('Import run'),
+            'url'   => 'director/list/importrun')
         );
         return $this->view->tabs;
     }